"""fileShare 스타일 로컬 웹 GUI — JUC500 USB bulk 브릿지.""" from __future__ import annotations import json import mimetypes import os import shutil import signal import socket import subprocess import sys import tempfile import threading import time import urllib.parse import webbrowser import atexit from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Optional from .clipboard_files import clipboard_preview, list_clipboard_files from .inputshare import SHARE from .protocol import Session, ROOTS_PATH, list_fs_roots from .usb_link import ( DeviceInfo, Juc500Link, list_devices, doctor, order_devices, data_link_devices, available_cable_models, CABLE_MODEL_JUC500, CABLE_MODEL_JUC700, CABLE_MODEL_SMARTKMLINK, KNOWN_CABLE_MODELS, PID_JUC700_MIN, PID_JUC700_MAX, _has_juc700_marker, ) from .wormhole_kill import kill_wormhole_processes, wormhole_status, kill_other_juc500_processes DEFAULT_GUI_PORT = 8765 _GUI_SERVER: ThreadingHTTPServer | None = None _QUIT_ONCE = threading.Event() _LAST_RESTART_AT = 0.0 _RESTART_COOLDOWN_S = 45.0 _autolisten_stop = threading.Event() _autolisten_thread: threading.Thread | None = None _autoconnect_stop = threading.Event() _autoconnect_thread: threading.Thread | None = None _last_autoconnect_attempt = 0.0 _connect_generation = 0 _connect_started_at = 0.0 _listen_opened_at = 0.0 _usb_setup_lock = threading.Lock() _CLEANUP_ONCE = threading.Event() _cable_watch_stop = threading.Event() _cable_watch_thread: threading.Thread | None = None _last_cable_swap_at = 0.0 CABLE_SWAP_COOLDOWN_S = 2.5 def restore_all_input(*, notify_peer: bool = True) -> None: """Always restore OS mouse/keyboard — safe to call multiple times.""" try: SHARE.ensure_local_input_restored(notify_peer=notify_peer) except Exception: try: from .inputshare import _force_release_os_input _force_release_os_input() except Exception: pass def _atexit_restore_input() -> None: if _CLEANUP_ONCE.is_set(): return _CLEANUP_ONCE.set() restore_all_input(notify_peer=True) def _project_root() -> Path: """Dev tree or PyInstaller EXE directory (start.bat / juc500_app.py 위치).""" env_root = os.environ.get("JUC500_ROOT", "").strip() if env_root: p = Path(env_root) if ( (p / "juc500_app.py").is_file() or (p / "start.bat").is_file() or (p / "start.sh").is_file() ): return p.resolve() candidates: list[Path] = [] if getattr(sys, "frozen", False): candidates.append(Path(sys.executable).resolve().parent) here = Path(__file__).resolve().parent candidates += [here.parent, Path.cwd()] # Windows dev default from sync_peer / start.bat if sys.platform == "win32": candidates += [ Path(r"C:\dev\juc500"), Path(r"C:\dev\juc500\juc500"), ] seen: set[str] = set() for base in candidates: p = base.resolve() for _ in range(7): key = str(p) if key in seen: break seen.add(key) if (p / "juc500_app.py").is_file() or (p / "start.bat").is_file(): return p if (p / "start.sh").is_file() and (p / "juc500_xfer").is_dir(): return p if p.parent == p: break p = p.parent return here.parent.resolve() def _win_detached_flags() -> int: return ( getattr(subprocess, "DETACHED_PROCESS", 0) | getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) ) def _spawn_delayed(cmd: list[str], *, cwd: str, env: dict[str, str] | None = None) -> None: """Give the current process time to release USB before the child starts.""" def _run() -> None: time.sleep(2.0) try: subprocess.Popen( cmd, cwd=cwd, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=_win_detached_flags() if sys.platform == "win32" else 0, start_new_session=sys.platform != "win32", close_fds=True, ) except Exception as exc: print(f"재시작 자식 프로세스 실패: {exc}", flush=True) threading.Thread(target=_run, name="juc500-restart-delay", daemon=True).start() def _windows_bash() -> str | None: """Git Bash / MSYS bash for start.sh on Windows.""" for name in ("bash", "bash.exe"): found = shutil.which(name) if found: return found for candidate in ( Path(r"C:\Program Files\Git\bin\bash.exe"), Path(r"C:\Program Files\Git\usr\bin\bash.exe"), ): if candidate.is_file(): return str(candidate) return None def _spawn_windows_restart(root: Path, log_file: Path) -> None: root = root.resolve() log_file.parent.mkdir(parents=True, exist_ok=True) root_s = str(root) env = os.environ.copy() env["PYTHONPATH"] = root_s + (";" + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") env["JUC500_ROOT"] = root_s start_sh = root / "start.sh" start_bat = root / "start.bat" py = root / ".venv" / "Scripts" / "python.exe" app = root / "juc500_app.py" if getattr(sys, "frozen", False): exe = Path(sys.executable).resolve() _spawn_delayed([str(exe)], cwd=str(exe.parent), env=env) return # Void / Git Bash dev flow — same entry as manual ./start.sh bash = _windows_bash() if bash and start_sh.is_file(): _spawn_delayed([bash, str(start_sh)], cwd=root_s, env=env) return if start_bat.is_file(): comspec = os.environ.get("COMSPEC", r"C:\Windows\System32\cmd.exe") if not Path(comspec).is_file(): comspec = "cmd.exe" _spawn_delayed( [comspec, "/c", f'ping 127.0.0.1 -n 3 >nul & call "{start_bat}"'], cwd=root_s, env=env, ) return if py.is_file() and app.is_file(): _spawn_delayed([str(py), str(app)], cwd=root_s, env=env) return if app.is_file(): _spawn_delayed([sys.executable, str(app)], cwd=root_s, env=env) return raise FileNotFoundError( f"재시작 스크립트 없음 ({root}) — start.sh / start.bat 또는 " ".venv\\Scripts\\python.exe + juc500_app.py 필요" ) def _spawn_unix_restart(root: Path, log_file: Path) -> None: root = root.resolve() log_file.parent.mkdir(parents=True, exist_ok=True) start_sh = root / "start.sh" if start_sh.is_file(): cmd = f"sleep 1.2; exec '{start_sh}' >>'{log_file}' 2>&1" else: py = root / ".venv" / "bin" / "python" app = root / "juc500_app.py" if not (py.is_file() and app.is_file()): raise FileNotFoundError(f"재시작 스크립트 없음 ({root})") cmd = ( f"sleep 1.2; cd '{root}' && " f"exec '{py}' '{app}' >>'{log_file}' 2>&1" ) subprocess.Popen( ["/bin/bash", "-c", cmd], cwd=str(root), start_new_session=True, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) def schedule_restart(reason: str = "재시작") -> None: """Spawn detached start script, then quit (local + RPC peer restart).""" global _LAST_RESTART_AT if _QUIT_ONCE.is_set(): return now = time.monotonic() if now - _LAST_RESTART_AT < _RESTART_COOLDOWN_S: msg = f"재시작 요청 무시 (쿨다운 {int(_RESTART_COOLDOWN_S)}s) — 수동 ./start.sh 사용" try: STATE.log(msg) except Exception: print(msg) return _LAST_RESTART_AT = now root = _project_root() log_file = (root / ".juc500" / "restart.log").resolve() try: STATE.log(reason) except Exception: print(reason) try: if sys.platform == "win32": _spawn_windows_restart(root, log_file) else: _spawn_unix_restart(root, log_file) except Exception as exc: msg = f"재시작 스크립트 실행 실패: {exc} (root={root})" try: STATE.log(msg) except Exception: print(msg) return request_quit(reason) def request_quit(reason: str = "") -> None: """Disable input share, disconnect USB, stop HTTP server, exit process.""" if _QUIT_ONCE.is_set(): return _QUIT_ONCE.set() msg = "Esc — 입력공유 끄고 종료" if not reason else reason try: STATE.log(msg) except Exception: print(msg) def _do() -> None: restore_all_input(notify_peer=True) try: disconnect_usb() except Exception: pass srv = _GUI_SERVER if srv is not None: try: srv.shutdown() except Exception: pass try: srv.server_close() except Exception: pass # Hard-exit so suppressed listeners / run loops cannot keep the process alive os._exit(0) threading.Thread(target=_do, name="juc500-quit", daemon=True).start() def _static_root() -> Path: """개발/PyInstaller frozen 모두에서 static 루트를 찾는다.""" import sys candidates: list[Path] = [] if getattr(sys, "frozen", False): mei = getattr(sys, "_MEIPASS", None) if mei: candidates += [ Path(mei) / "static", Path(mei) / "juc500_xfer" / "static", ] candidates.append(Path(sys.executable).resolve().parent / "static") here = Path(__file__).resolve().parent candidates += [here / "static", here.parent / "static"] for c in candidates: if (c / "index.html").is_file(): return c.resolve() return (here / "static").resolve() def _static_file(rel: str) -> Path | None: root = _static_root() name = (rel or "").lstrip("/").replace("\\", "/") if not name or ".." in name.split("/"): return None target = (root / name).resolve() try: target.relative_to(root) except ValueError: return None return target if target.is_file() else None class GuiState: def __init__(self) -> None: self.lock = threading.Lock() self.link: Juc500Link | None = None self.session: Session | None = None self.peer_name: str | None = None self.local_root = Path.home().resolve() self.serial: str | None = None env_model = os.environ.get("JUC500_CABLE_MODEL", "").strip() # Normalize SmartKMLink casing from env env_upper = env_model.upper().replace(" ", "") if env_upper in ("JUC500", "JUC700"): self.cable_model = env_upper elif env_upper in ("SMARTKMLINK", "KMLINK"): self.cable_model = CABLE_MODEL_SMARTKMLINK else: self.cable_model = CABLE_MODEL_JUC700 # Mac↔Win JUC700 path is primary now self.active_device: DeviceInfo | None = None self.gui_port = DEFAULT_GUI_PORT self.auto_status = "JUC500/JUC700 케이블을 연결하세요" self.last_progress = {"done": 0, "total": 0, "label": "", "error": None, "busy": False} self.connecting = False self.connect_error: str | None = None self.listening = False # USB open, waiting for peer HELLO (auto-connect) self.usb_finalized = False # SHARE bound + handshake UI complete self.usb_role = "unknown" # connector | listener — connector pushes sources to peer self.log_lines: list[str] = [] self.auto_kill_wormhole = True # Clipboard paste → remote folder (last browsed remote path in UI) self.paste_dest: str = "" self.clipboard_sending = False self.clipboard_send_error: str | None = None def log(self, msg: str) -> None: self.log_lines.append(msg) self.log_lines = self.log_lines[-200:] print(msg) def set_progress(self, done: int, total: int, label: str) -> None: self.last_progress = { "done": done, "total": total, "label": label, "error": None, "busy": total > 0 and done < total, } STATE = GuiState() SHARE._log = STATE.log def _peer_auto_sync_enabled() -> bool: """Dev-only: push sources to peer on connect (see scripts/sync_peer.sh).""" return os.environ.get("JUC500_AUTO_SYNC_PEER", "0").strip().lower() in ( "1", "true", "yes", "on", ) def _app_version_info() -> dict[str, str]: try: from . import __version__ as pkg_version except Exception: pkg_version = "unknown" try: from .inputshare import INPUTSHARE_BUILD as inputshare_build except Exception: inputshare_build = "unknown" try: from .usb_link import XFER_BUILD as xfer_build except Exception: xfer_build = "unknown" return { "version": str(pkg_version), "inputshare_build": inputshare_build, "xfer_build": xfer_build, } def _clipboard_status() -> dict[str, Any]: preview = clipboard_preview() return { **preview, "paste_dest": STATE.paste_dest, "sending": STATE.clipboard_sending, "error": STATE.clipboard_send_error, "ok": True, } def _resolve_paste_dest() -> str: dest = _norm_peer_dest(STATE.paste_dest) if dest and dest not in (ROOTS_PATH, ".", "./"): return dest # Fallback: peer Desktop relative name — UI should normally set paste_dest return "Desktop" def start_clipboard_send(*, source: str = "api") -> dict[str, Any]: """Send clipboard files to STATE.paste_dest over the USB session.""" if not _session_alive(): return {"ok": False, "error": "먼저 USB 「연결」을 완료하세요."} if STATE.clipboard_sending: return {"ok": False, "error": "이미 클립보드 파일을 전송 중입니다.", "sending": True} files = list_clipboard_files() if not files: return {"ok": False, "error": "클립보드에 파일이 없습니다. Finder/Explorer에서 복사하세요."} dest = _resolve_paste_dest() paths = [str(p) for p in files] def _run() -> None: STATE.clipboard_sending = True STATE.clipboard_send_error = None try: STATE.log( f"클립보드 붙여넣기 → {dest} ({len(paths)}개, src={source}): " + ", ".join(p.name for p in files[:5]) + ("…" if len(files) > 5 else "") ) result = _send_local_paths(paths, dest) STATE.log(f"클립보드 전송 완료: {result.get('count')}개 → {dest}") except Exception as exc: STATE.clipboard_send_error = str(exc) STATE.log(f"클립보드 전송 실패: {exc}") STATE.last_progress = { "done": 0, "total": 1, "label": "클립보드 전송", "error": str(exc), "busy": False, } finally: STATE.clipboard_sending = False threading.Thread(target=_run, name="juc500-clipboard-send", daemon=True).start() return { "ok": True, "started": True, "count": len(paths), "dest_dir": dest, "names": [p.name for p in files[:8]], } # InputShare paste intercept → clipboard send SHARE.on_clipboard_paste = lambda: start_clipboard_send(source="inputshare") def _session_alive() -> bool: return STATE.session is not None and STATE.session._connected.is_set() def _resolve_fs_path(path_str: str) -> Path: """Browse path: empty → home (local_root); absolute / Drive paths allowed (parents OK).""" raw = (path_str or "").replace("\\", "/").strip() if not raw or raw in (".", "./"): return STATE.local_root.resolve() # Absolute POSIX or Windows (C:/..., //server/...) if raw.startswith("/") or raw.startswith("//") or (len(raw) >= 2 and raw[1] == ":"): return Path(raw).resolve() # Legacy relative-under-home return (STATE.local_root / raw.lstrip("/")).resolve() def _path_to_api(path: Path) -> str: s = path.resolve().as_posix() # Normalize Windows drive root to "C:/" if len(s) == 2 and s[1] == ":": s += "/" return s def _safe_local(rel: str) -> Path: return _resolve_fs_path(rel) def _list_local_entries(rel: str) -> dict[str, Any]: raw = (rel or "").replace("\\", "/").strip() if raw == ROOTS_PATH: return list_fs_roots() path = _resolve_fs_path(rel) if not path.exists(): raise FileNotFoundError(rel or str(STATE.local_root)) if not path.is_dir(): raise NotADirectoryError(rel) entries = [] for child in sorted(path.iterdir(), key=lambda p: (not p.is_dir(), p.name.lower())): try: st = child.stat() entries.append( { "name": child.name, "is_dir": child.is_dir(), "size": 0 if child.is_dir() else int(st.st_size), "mtime": int(st.st_mtime), } ) except OSError: continue return {"path": _path_to_api(path), "entries": entries} def _device_dict(d: DeviceInfo) -> dict[str, Any]: return { "bus": d.bus, "address": d.address, "serial": d.serial, "product": d.product, "model": d.model, "vid": f"{d.vid:04x}", "pid": f"{d.pid:04x}", "if_num": d.if_num, } def _ordered_cables() -> list[DeviceInfo]: return order_devices( list_devices(), serial=STATE.serial, model=STATE.cable_model, ) def _apply_cable_availability(devices: list[DeviceInfo]) -> tuple[str, ...]: """Refresh which cable models are plugged in; auto-select when only one.""" available = available_cable_models(devices) with STATE.lock: if len(available) == 1: STATE.cable_model = available[0] elif available and STATE.cable_model not in available: if CABLE_MODEL_JUC500 in available: STATE.cable_model = CABLE_MODEL_JUC500 else: STATE.cable_model = available[0] return available def _cables_for_connect() -> list[DeviceInfo]: """Pick USB nodes for file transfer (always Smart Data Link IF5, never JUC700 DSS IF0).""" ordered = _ordered_cables() links = data_link_devices(ordered) if not links: return [] model = STATE.cable_model if model == CABLE_MODEL_JUC500: preferred = [d for d in links if d.model == CABLE_MODEL_JUC500] return preferred or links if model == CABLE_MODEL_JUC700: preferred = [d for d in links if d.model == CABLE_MODEL_JUC700] return preferred or links if model == CABLE_MODEL_SMARTKMLINK: preferred = [d for d in links if d.model == CABLE_MODEL_SMARTKMLINK] return preferred or links return links def _device_locators(devices: list[DeviceInfo]) -> set[tuple[int, int]]: return {(int(d.bus), int(d.address)) for d in devices} def _active_still_present( devices: list[DeviceInfo], active: DeviceInfo | None ) -> bool: if active is None: return True return (int(active.bus), int(active.address)) in _device_locators(devices) def _cable_hotplug_reason(devices: list[DeviceInfo]) -> str | None: """Return a reconnect reason when the plugged cable set no longer matches USB state.""" with STATE.lock: prev_model = STATE.cable_model active = STATE.active_device alive = _session_alive() listening = STATE.listening connecting = STATE.connecting if connecting: return None available = _apply_cable_availability(devices) wanted = _cables_for_connect() usb_busy = alive or listening if not usb_busy: return None if not devices: return "케이블 분리 감지" if not wanted: if any(PID_JUC700_MIN <= int(d.pid) <= PID_JUC700_MAX for d in devices): # Marker present but no usable data-link node yet (enumeration lag). return None return "케이블 분리 감지" if active is not None and not _active_still_present(devices, active): model = STATE.cable_model or (wanted[0].model if wanted else "케이블") return f"케이블 교체 → {model}" wanted_locs = _device_locators(wanted) if active is not None and (int(active.bus), int(active.address)) not in wanted_locs: model = STATE.cable_model or wanted[0].model return f"케이블 교체 → {model}" if ( len(available) == 1 and prev_model != available[0] and wanted and (active is not None or alive or listening) ): return f"케이블 자동 전환 → {available[0]}" with STATE.lock: new_model = STATE.cable_model if ( new_model != prev_model and wanted and (alive or listening) and new_model in available ): return f"케이블 자동 전환 → {new_model}" return None def _handle_cable_hotplug(reason: str) -> None: """Tear down stale USB and reopen on the newly plugged cable/model.""" global _last_cable_swap_at now = time.monotonic() if now - _last_cable_swap_at < CABLE_SWAP_COOLDOWN_S: return if not _cable_switch_lock.acquire(blocking=False): return try: with STATE.lock: if not (_session_alive() or STATE.listening): return _last_cable_swap_at = now STATE.log(f"{reason} — USB 재연결 중…") with STATE.lock: STATE.auto_status = f"{reason} — 재연결 중…" disconnect_usb(restart_listen=True) finally: _cable_switch_lock.release() def _cable_hotplug_loop() -> None: """Detect unplug/swap (JUC500↔JUC700) while connected and trigger auto-reconnect.""" while not _cable_watch_stop.is_set(): try: devices = list_devices() except Exception: devices = [] reason = _cable_hotplug_reason(devices) if reason: _handle_cable_hotplug(reason) _cable_watch_stop.wait(3.0) continue _cable_watch_stop.wait(1.0) def start_cable_watch() -> None: global _cable_watch_thread _cable_watch_stop.clear() if _cable_watch_thread is not None and _cable_watch_thread.is_alive(): return _cable_watch_thread = threading.Thread( target=_cable_hotplug_loop, name="juc500-cable-watch", daemon=True ) _cable_watch_thread.start() def stop_cable_watch() -> None: _cable_watch_stop.set() def _prepare_usb_before_open() -> None: if not STATE.auto_kill_wormhole: return STATE.auto_status = "공식 Wormhole 종료 중…" STATE.log("공식 Wormhole 프로세스 종료 시도") result = kill_wormhole_processes(force=True) STATE.log(result.get("message") or "") if result.get("killed"): time.sleep(0.4) def _on_usb_link_broken() -> None: """USB bulk Entity not found — stale IF5 handle; re-enter listen mode.""" STATE.log("USB 링크 끊김 감지 — 대기 모드 재진입") _close_listen_session() disconnect_usb(restart_listen=True) def _open_usb_link_for( dev: DeviceInfo, *, prepare: bool = True, _retry: bool = True ) -> tuple[Juc500Link, Session]: """Open a specific cable by bus/address.""" if prepare: _prepare_usb_before_open() try: link = Juc500Link( bus=dev.bus, address=dev.address, serial=STATE.serial, model=dev.model or STATE.cable_model, ) except RuntimeError as exc: if _retry and "IF5" in str(exc): fresh = _cables_for_connect() if fresh: alt = fresh[0] if (alt.bus, alt.address) != (dev.bus, dev.address): STATE.log( f"USB 주소 갱신 — bus={dev.bus} addr={dev.address} " f"→ bus={alt.bus} addr={alt.address}" ) return _open_usb_link_for(alt, prepare=False, _retry=False) raise session = Session( link, on_log=STATE.log, local_root=STATE.local_root, ) session.inbound_dest = STATE.local_root session.on_progress = STATE.set_progress session.on_connected = _on_peer_connected session.on_restart = lambda: schedule_restart("상대 요청 — 재시작") session.on_cable_switch = _on_peer_cable_switch session.on_link_broken = _on_usb_link_broken session.start() return link, session def _open_usb_link() -> tuple[Juc500Link, Session]: """Open the preferred cable (first in priority order).""" devices = _ordered_cables() if not devices: raise RuntimeError("JUC500/JUC700(Smart Data Link) 장치가 없습니다. 케이블을 연결하세요.") dev = devices[0] link, session = _open_usb_link_for(dev) with STATE.lock: STATE.active_device = dev STATE.log( f"{dev.model} 케이블 선택 (IF{dev.if_num}, {dev.vid:04x}:{dev.pid:04x}" f"{', ' + dev.serial if dev.serial else ''})" ) return link, session def _finalize_usb_session(session: Session, link: Juc500Link, peer: str) -> bool: """Bind session after HELLO handshake. Idempotent.""" with STATE.lock: if STATE.session is not session: return False if STATE.usb_finalized: return False STATE.usb_finalized = True STATE.link = link STATE.session = session STATE.peer_name = peer STATE.auto_status = f"연결됨 · {peer}" STATE.connecting = False STATE.listening = False STATE.connect_error = None time.sleep(1.8) try: link.drain(0.5) except Exception: pass SHARE.bind_session(session) SHARE._log = STATE.log _auto_enable_inputshare_on_connect() threading.Thread( target=_sync_cable_model_to_peer, name="juc500-cable-sync", daemon=True, ).start() if _peer_auto_sync_enabled(): threading.Thread( target=_auto_sync_sources_to_peer, name="juc500-peer-source-sync", daemon=True, ).start() return True def _auto_sync_sources_to_peer() -> None: """Connector side: push local sources to peer after USB session is ready (dev only).""" time.sleep(0.6) with STATE.lock: if STATE.usb_role != "connector": return session = STATE.session if session is None or not _session_alive(): return try: from .peer_sync import sync_sources_to_peer sync_sources_to_peer(session, on_log=STATE.log, on_progress=STATE.set_progress) except Exception as exc: STATE.log(f"상대 소스 자동 동기화 실패: {exc}") finally: STATE.last_progress = { "done": 1, "total": 1, "label": "완료", "error": None, "busy": False, } def _sync_cable_model_to_peer() -> None: with STATE.lock: model = STATE.cable_model session = STATE.session if model not in KNOWN_CABLE_MODELS: return if session is None or not _session_alive(): return try: session.rpc("set_cable_model", model=model, timeout=8.0) except Exception as exc: STATE.log(f"상대 케이블 동기화: {exc}") def _auto_enable_inputshare_on_connect() -> None: """Default input share ON when USB session is ready; syncs to peer via SHARE_ON.""" info = SHARE.set_enabled(True) if info.get("ok"): STATE.log("입력공유 켜짐 — 연결 시 기본 ON (한쪽에서 끄면 상대도 꺼짐)") else: err = (info.get("error") or "권한 필요").strip() STATE.log( f"입력공유 자동 켜기 실패: {err} — " "손쉬운 사용 허용 후 「입력공유」를 다시 눌러 주세요" ) def _recover_usb_access() -> None: """Release IF5/IF0 when another process or stale session blocks USB claim.""" if STATE.auto_kill_wormhole: try: kill_wormhole_processes(force=True) except Exception: pass try: result = kill_other_juc500_processes(force=True) msg = (result.get("message") or "").strip() if msg: STATE.log(msg) except Exception: pass with STATE.lock: session = STATE.session link = STATE.link if _session_alive(): return STATE.session = None STATE.link = None STATE.listening = False if session is not None: try: session.stop() except Exception: pass if link is not None: try: link.close() except Exception: pass def _ensure_usb_finalized() -> bool: """Complete GUI bind when USB HELLO finished but finalize was missed (race).""" with STATE.lock: session = STATE.session link = STATE.link if session is None or link is None: return False if STATE.usb_finalized: return True if not session._connected.is_set(): return False peer = session._peer_hello or STATE.peer_name or "peer" if _finalize_usb_session(session, link, peer): STATE.log(f"상대 연결 감지 — 자동 연결 완료 ({peer})") return True return False def _on_peer_connected(peer: str) -> None: """Passive or active: HELLO/HELLO_ACK received — finalize without pressing 「연결」.""" with STATE.lock: session = STATE.session link = STATE.link auto = STATE.listening and not STATE.connecting if session is None or link is None: return if auto: STATE.log(f"상대 HELLO 수신 — 연결 마무리 중 ({peer})") _ensure_usb_finalized() def _close_listen_session() -> None: """Tear down a listen session that never completed handshake.""" with STATE.lock: session = STATE.session link = STATE.link if _session_alive(): return STATE.session = None STATE.link = None STATE.listening = False STATE.usb_finalized = False if session is not None: try: session.stop() except Exception: pass if link is not None: try: link.close() except Exception: pass def _autolisten_loop() -> None: """Keep USB open when idle so the peer can connect with one button press.""" while not _autolisten_stop.is_set(): if _session_alive(): _autolisten_stop.wait(1.0) continue with STATE.lock: if STATE.connecting or STATE.listening: _autolisten_stop.wait(0.8) continue try: devices = list_devices() _apply_cable_availability(devices) except Exception as exc: with STATE.lock: if not _session_alive() and not STATE.connecting and not STATE.listening: STATE.auto_status = f"USB 오류: {exc}" _autolisten_stop.wait(3.0) continue cables = _cables_for_connect() if not cables: with STATE.lock: if ( not _session_alive() and not STATE.connecting and not STATE.listening ): model = STATE.cable_model or "선택한" if _has_juc700_marker(devices) and model == CABLE_MODEL_JUC700: STATE.auto_status = ( "JUC700 감지 — 파일전송 노드 대기 중 " "(Mac↔Win: 양쪽 IF0 · Win은 MI_00 WinUSB)" ) elif devices: STATE.auto_status = ( f"{model} 케이블을 선택했지만 연결된 장치가 없습니다 " f"(감지: {len(devices)}개)" ) else: STATE.auto_status = "JUC500/JUC700 케이블을 연결하세요" _autolisten_stop.wait(2.0) continue opened = False _prepare_usb_before_open() for dev in cables: try: with _usb_setup_lock: with STATE.lock: if _session_alive() or STATE.connecting or STATE.listening: opened = True break link, session = _open_usb_link_for(dev, prepare=False) with STATE.lock: STATE.link = link STATE.session = session STATE.active_device = dev STATE.listening = True STATE.usb_role = "listener" global _listen_opened_at _listen_opened_at = time.monotonic() n = len(cables) extra = f" ({n}개 감지)" if n > 1 else "" STATE.auto_status = ( f"{dev.model} 케이블 감지{extra} — 상대 연결 대기 중…" ) ifn = link.layout.if_num display_model = dev.model or link.model STATE.log( f"USB 대기 모드 — {display_model} IF{ifn} " f"(Smart Data Link · 상대 실행 시 자동 연결)" ) opened = True break except Exception as exc: err = str(exc) STATE.log(f"{dev.model} 대기 모드 열기 실패: {exc}") if ( dev.model == CABLE_MODEL_JUC700 or "not supported" in err.lower() or "지원하지 않" in err ): with STATE.lock: if not _session_alive() and not STATE.connecting: STATE.auto_status = ( "JUC700 — Zadig: Smart Data Link → Interface 5 (MI_05) → WinUSB" ) if "Access denied" in err or "점유" in err: _recover_usb_access() if not opened: with STATE.lock: if not _session_alive() and not STATE.connecting: STATE.auto_status = ( "케이블 열기 실패 — python -m juc500_xfer doctor" ) _autolisten_stop.wait(4.0) continue while not _autolisten_stop.is_set(): need_finalize = False with STATE.lock: if _session_alive(): need_finalize = not STATE.usb_finalized break if not STATE.listening or STATE.session is None: break if STATE.connecting: break if need_finalize: _ensure_usb_finalized() _autolisten_stop.wait(0.5) with STATE.lock: stale = ( STATE.listening and STATE.session is not None and not _session_alive() and not STATE.connecting ) if stale: _close_listen_session() _autolisten_stop.wait(1.0) def start_autolisten() -> None: """Start background USB listen thread (idempotent).""" global _autolisten_thread _autolisten_stop.clear() if _autolisten_thread is not None and _autolisten_thread.is_alive(): return _autolisten_thread = threading.Thread( target=_autolisten_loop, name="juc500-autolisten", daemon=True ) _autolisten_thread.start() def stop_autolisten() -> None: _autolisten_stop.set() def _abort_stale_connect(*, reason: str) -> None: """Drop a long-running active connect so autolisten can open USB.""" global _connect_generation with STATE.lock: if not STATE.connecting: return session = STATE.session link = STATE.link _connect_generation += 1 STATE.connecting = False STATE.connect_error = None STATE.listening = False if not _session_alive(): STATE.session = None STATE.link = None if session is not None and not _session_alive(): try: session.stop() except Exception: pass if link is not None and not _session_alive(): try: link.close() except Exception: pass STATE.log(reason) start_autolisten() def _autoconnect_loop() -> None: """Keep USB listening and finalize when peer completes handshake.""" global _last_autoconnect_attempt while not _autoconnect_stop.is_set(): try: if _ensure_usb_finalized(): _autoconnect_stop.wait(1.0) continue with STATE.lock: if STATE.usb_finalized and _session_alive(): _autoconnect_stop.wait(1.0) continue listening = STATE.listening handshake = _session_alive() and not STATE.usb_finalized connecting = STATE.connecting err = STATE.connect_error or "" listen_age = ( time.monotonic() - _listen_opened_at if listening else 0.0 ) connect_age = ( time.monotonic() - _connect_started_at if connecting else 0.0 ) if handshake: _ensure_usb_finalized() _autoconnect_stop.wait(0.8) continue if connecting and connect_age > 90.0: _abort_stale_connect( reason="능동 연결 지연 — USB 대기 모드로 전환" ) _autoconnect_stop.wait(2.0) continue if connecting: _autoconnect_stop.wait(0.6) continue if listening: _ensure_usb_finalized() with STATE.lock: session = STATE.session active = STATE.active_device link = STATE.link # Prefer live link layout — DeviceInfo.if_num==0 must not use # `x or 5` (0 is falsy and was mis-classified as IF5). layout_if = None if link is not None: try: layout_if = int(link.layout.if_num) except Exception: layout_if = None if layout_if is None and active is not None: raw_if = getattr(active, "if_num", None) layout_if = 5 if raw_if is None else int(raw_if) model = None if active is not None: model = active.model elif link is not None: try: model = link.model except Exception: model = None is_juc700_if0 = bool( model == CABLE_MODEL_JUC700 and layout_if == 0 ) if session is not None: try: session.poke_hello() except Exception: pass # IF5 only (JUC500/SmartKMLink): Mac becomes connector. # JUC700 IF0: stay listener + poke (reopen → Access denied). if ( not is_juc700_if0 and sys.platform != "win32" and listen_age > 2.5 and not STATE.connecting ): _last_autoconnect_attempt = time.monotonic() STATE.log( "IF5 대기 → 능동 연결 (Mac=connector / Win=listener)" ) start_connect_usb_async(timeout=60.0) _autoconnect_stop.wait(1.0) continue if listen_age > 8.0 and not is_juc700_if0: _last_autoconnect_attempt = time.monotonic() start_connect_usb_async(timeout=60.0) _autoconnect_stop.wait(1.0) continue if not _cables_for_connect(): _autoconnect_stop.wait(2.0) continue if "Access denied" in err or "점유" in err: STATE.log("USB 점유 충돌 — 잠시 대기 후 재시도") _recover_usb_access() with STATE.lock: STATE.connect_error = None _autoconnect_stop.wait(5.0) start_autolisten() _last_autoconnect_attempt = time.monotonic() _autoconnect_stop.wait(2.0) continue start_autolisten() _autoconnect_stop.wait(1.5) except Exception as exc: STATE.log(f"자동 연결 오류: {exc}") _autoconnect_stop.wait(1.0) def start_autoconnect() -> None: """Background auto-connect (idempotent).""" global _autoconnect_thread _autoconnect_stop.clear() if _autoconnect_thread is not None and _autoconnect_thread.is_alive(): return _autoconnect_thread = threading.Thread( target=_autoconnect_loop, name="juc500-autoconnect", daemon=True ) _autoconnect_thread.start() def stop_autoconnect() -> None: _autoconnect_stop.set() def _arm_connect_watchdog(gen: int, seconds: float) -> None: """Force-clear stuck connecting flag if USB handshake hangs.""" def _watch() -> None: time.sleep(seconds + 5.0) with STATE.lock: if gen != _connect_generation or not STATE.connecting: return session = STATE.session STATE.connecting = False STATE.connect_error = "연결 시간 초과" STATE.auto_status = "연결 시간 초과 — 다시 시도하세요" if session is not None: try: session.stop() except Exception: pass STATE.log("연결 시간 초과 — 감시자가 세션을 종료했습니다") threading.Thread( target=_watch, name="juc500-connect-watchdog", daemon=True ).start() def connect_usb(timeout: float = 45.0) -> dict[str, Any]: with STATE.lock: if _session_alive(): if not STATE.usb_finalized: session = STATE.session link = STATE.link else: session = link = None if session is not None and link is not None: pass # finalize below outside lock else: return { "ok": True, "host": STATE.peer_name or "peer", "hostname": STATE.peer_name or "peer", "port": 0, "start_path": "", "receive_path": "", } else: session = link = None if session is None: if STATE.connecting: raise RuntimeError("이미 연결 중입니다") my_gen = _connect_generation global _connect_started_at _connect_started_at = time.monotonic() STATE.connecting = True STATE.connect_error = None STATE.usb_role = "connector" reuse_listen = STATE.listening and STATE.session is not None STATE.auto_status = ( "상대 프로그램 연결 중…" if reuse_listen else "USB 링크 여는 중…" ) else: my_gen = _connect_generation reuse_listen = False if session is not None and link is not None: _ensure_usb_finalized() with STATE.lock: return { "ok": True, "host": STATE.peer_name or "peer", "hostname": STATE.peer_name or "peer", "port": 0, "start_path": "", "receive_path": "", } _arm_connect_watchdog(my_gen, timeout) link: Juc500Link | None = None session: Session | None = None try: if reuse_listen: with STATE.lock: link = STATE.link session = STATE.session STATE.listening = False if link is None or session is None: raise RuntimeError("USB 세션을 열 수 없습니다") if session._connected.is_set(): _ensure_usb_finalized() with STATE.lock: peer = STATE.peer_name or session._peer_hello or "peer" model = STATE.active_device.model if STATE.active_device else "케이블" STATE.connecting = False STATE.log(f"연결됨 — {peer} ({model})") return { "ok": True, "host": peer, "hostname": peer, "port": 0, "start_path": "", "receive_path": "", "cable_model": model, } STATE.auto_status = "상대 프로그램 대기 중…" peer = session.wait_peer(timeout=timeout) _ensure_usb_finalized() or _finalize_usb_session(session, link, peer) model = STATE.active_device.model if STATE.active_device else "케이블" STATE.log(f"연결됨 — {peer} ({model})") return { "ok": True, "host": peer, "hostname": peer, "port": 0, "start_path": "", "receive_path": "", "cable_model": model, } else: devices = _cables_for_connect() if not devices: model = STATE.cable_model or "선택한" all_devs = _ordered_cables() if all_devs: raise RuntimeError( f"{model} 케이블이 선택되었지만 해당 장치가 없습니다. " f"상단에서 케이블 종류를 바꾸거나 USB를 확인하세요." ) raise RuntimeError( "JUC500/JUC700(Smart Data Link) 장치가 없습니다. 케이블을 연결하세요." ) per_timeout = max(12.0, timeout / max(1, len(devices))) last_err: Exception | None = None with STATE.lock: prev_err = STATE.connect_error or "" if "Access denied" in prev_err or "점유" in prev_err: _recover_usb_access() _prepare_usb_before_open() for dev in devices: try: with _usb_setup_lock: link, session = _open_usb_link_for(dev, prepare=False) with STATE.lock: STATE.link = link STATE.session = session STATE.active_device = dev STATE.log( f"{dev.model} 케이블 시도 (IF{dev.if_num}, " f"{dev.vid:04x}:{dev.pid:04x})" ) STATE.auto_status = ( f"{dev.model} — 상대에 HELLO 송신 중… " "(상대도 이 앱 실행·대기 모드여야 연결 감지)" ) peer = session.wait_peer(timeout=per_timeout) _ensure_usb_finalized() or _finalize_usb_session(session, link, peer) STATE.log(f"연결됨 — {peer} ({dev.model})") return { "ok": True, "host": peer, "hostname": peer, "port": 0, "start_path": "", "receive_path": "", "cable_model": dev.model, } except Exception as exc: last_err = exc STATE.log(f"{dev.model} 연결 실패: {exc}") try: if session is not None: session.stop() except Exception: pass try: if link is not None: link.close() except Exception: pass link = None session = None with STATE.lock: STATE.link = None STATE.session = None STATE.active_device = None if last_err is not None: raise last_err raise RuntimeError("연결 가능한 케이블이 없습니다") except Exception as exc: if _session_alive() and _ensure_usb_finalized(): with STATE.lock: if my_gen == _connect_generation: STATE.connecting = False STATE.connect_error = None with STATE.lock: peer = STATE.peer_name or "peer" return { "ok": True, "host": peer, "hostname": peer, "port": 0, "start_path": "", "receive_path": "", } SHARE.on_disconnect() with STATE.lock: if my_gen == _connect_generation: STATE.connecting = False STATE.connect_error = str(exc).strip() or type(exc).__name__ if not _session_alive(): STATE.auto_status = "연결 실패 — 다시 시도하세요" if my_gen == _connect_generation and not _session_alive(): msg = str(exc) if "Access denied" in msg or "점유" in msg: _recover_usb_access() _close_listen_session() start_autolisten() raise finally: with STATE.lock: if my_gen == _connect_generation and STATE.connecting and not _session_alive(): STATE.connecting = False def start_connect_usb_async(timeout: float = 60.0) -> dict[str, Any]: """Start connect in a background thread so the HTTP request does not block. Browser fetch often aborts long /api/connect POSTs (~45s wait_peer), leaving the UI stuck on '연결 중…' even after HELLO succeeds. Poll /api/status instead. """ with STATE.lock: if _session_alive(): if not STATE.usb_finalized: _ensure_usb_finalized() return { "ok": True, "started": False, "already": True, "host": STATE.peer_name or "peer", "hostname": STATE.peer_name or "peer", } if STATE.connecting: return {"ok": True, "started": False, "pending": True} def _run() -> None: try: connect_usb(timeout=timeout) except Exception as exc: STATE.log(f"연결 실패: {exc}") threading.Thread(target=_run, daemon=True, name="juc500-connect").start() # Wait until connecting flag is set (or already connected) so status is consistent. for _ in range(50): with STATE.lock: if STATE.connecting or _session_alive() or STATE.connect_error: break time.sleep(0.02) return {"ok": True, "started": True, "async": True} def disconnect_usb(*, restart_listen: bool = True) -> None: global _connect_generation stop_autoconnect() with STATE.lock: _connect_generation += 1 session = STATE.session link = STATE.link STATE.session = None STATE.link = None STATE.peer_name = None STATE.paste_dest = "" STATE.clipboard_sending = False STATE.clipboard_send_error = None STATE.auto_status = "연결 해제됨" STATE.connecting = False STATE.listening = False STATE.usb_finalized = False STATE.usb_role = "unknown" STATE.active_device = None SHARE.on_disconnect() if session: try: session.stop() except Exception: pass if link: try: link.close() except Exception: pass if restart_listen: start_autolisten() start_autoconnect() _cable_switch_lock = threading.Lock() def switch_cable_model(model: str, *, from_peer: bool = False) -> dict[str, Any]: """Select cable model, tear down USB, and reconnect on that cable only.""" raw = model.strip() key = raw.upper().replace(" ", "") if key in ("SMARTKMLINK", "KMLINK"): model = CABLE_MODEL_SMARTKMLINK elif key in ("JUC500", "JUC700"): model = key else: raise ValueError(f"지원 케이블: {', '.join(KNOWN_CABLE_MODELS)}") with _cable_switch_lock: with STATE.lock: prev = STATE.cable_model STATE.cable_model = model if from_peer and prev == model: return {"ok": True, "unchanged": True, "model": model} STATE.log( f"케이블 선택 → {model}" + (" (상대 동기화)" if from_peer else " — 상대에 동기화") ) if not from_peer: with STATE.lock: session = STATE.session if _session_alive() else None if session is not None: try: session.rpc("set_cable_model", model=model, timeout=8.0) except Exception as exc: STATE.log(f"상대 케이블 동기화 실패: {exc}") stop_autolisten() if STATE.auto_kill_wormhole: try: kill_wormhole_processes(force=True) except Exception: pass disconnect_usb(restart_listen=False) _close_listen_session() _recover_usb_access() global _connect_generation with STATE.lock: _connect_generation += 1 STATE.connect_error = None STATE.connecting = False STATE.auto_status = f"{model} 케이블로 재연결 중…" start_autoconnect() initiate_active = not from_peer if initiate_active: def _delayed_active_connect() -> None: # Let peer finish USB teardown and open listen on the new interface first. time.sleep(0.9) start_connect_usb_async(timeout=60.0) threading.Thread( target=_delayed_active_connect, name="juc500-cable-reconnect", daemon=True, ).start() else: start_autolisten() return {"ok": True, "model": model, "reconnecting": True, "from_peer": from_peer} def _on_peer_cable_switch(model: str) -> None: try: switch_cable_model(model, from_peer=True) except Exception as exc: STATE.log(f"상대 케이블 전환 실패: {exc}") def _require_session() -> Session: if not STATE.session or not STATE.session._connected.is_set(): raise RuntimeError("연결되지 않았습니다") return STATE.session def _norm_peer_dest(dest_dir: str) -> str: """Preserve absolute POSIX/Windows dest paths; only strip relative junk.""" dest = (dest_dir or "").replace("\\", "/").strip() if not dest or dest in (".", "./"): return "" if dest.startswith("/") or dest.startswith("//") or (len(dest) >= 2 and dest[1] == ":"): # Keep absolute form; collapse trailing slashes except drive root "C:/" if len(dest) == 2 and dest[1] == ":": return dest + "/" if len(dest) == 3 and dest[1] == ":" and dest[2] == "/": return dest return dest.rstrip("/") or dest return dest.strip("/") def _join_peer_path(dest: str, name: str) -> str: name = name.replace("\\", "/").lstrip("/") if not dest: return name if dest.endswith("/"): return f"{dest}{name}" return f"{dest}/{name}" def _peer_parent(remote_name: str) -> str: p = remote_name.replace("\\", "/").rstrip("/") if not p: return "" if len(p) == 2 and p[1] == ":": return "" idx = p.rfind("/") if idx < 0: return "" if idx == 0: return "/" parent = p[:idx] if len(parent) == 2 and parent[1] == ":": return parent + "/" return parent def _send_local_paths(paths: list[str], dest_dir: str) -> dict[str, Any]: session = _require_session() dest = _norm_peer_dest(dest_dir) count = 0 def ensure_remote_dir(rel_dir: str) -> None: rel_dir = _norm_peer_dest(rel_dir) if not rel_dir or rel_dir in (".", "./"): return try: session.rpc("mkdir", path=rel_dir, timeout=15) except Exception: pass for rel in paths: local = _safe_local(rel) if local.is_file(): remote_name = _join_peer_path(dest, local.name) if dest else local.name parent = _peer_parent(remote_name) if parent: ensure_remote_dir(parent) STATE.set_progress(0, local.stat().st_size, local.name) session.on_progress = lambda done, total, name=local.name: STATE.set_progress(done, total, name) try: # Large files need more time for peer ACCEPT on half-duplex USB. timeout = 180.0 if local.stat().st_size > 32 * 1024 * 1024 else 120.0 session.send_file(local, remote_name=remote_name, accept_timeout=timeout) except Exception: STATE.last_progress = { "done": 0, "total": local.stat().st_size, "label": local.name, "error": "전송 실패", "busy": False, } raise count += 1 elif local.is_dir(): base = _join_peer_path(dest, local.name) if dest else local.name ensure_remote_dir(base) for file_path in local.rglob("*"): if not file_path.is_file(): continue rel_under = file_path.relative_to(local).as_posix() remote_name = _join_peer_path(base, rel_under) parent = _peer_parent(remote_name) if parent: ensure_remote_dir(parent) STATE.set_progress(0, file_path.stat().st_size, remote_name) session.on_progress = lambda done, total, name=remote_name: STATE.set_progress( done, total, name ) session.send_file(file_path, remote_name=remote_name) count += 1 else: raise FileNotFoundError(rel) STATE.last_progress = {"done": 1, "total": 1, "label": "완료", "error": None, "busy": False} return {"count": count, "dest_dir": dest} def _fetch_remote_paths(paths: list[str], dest_dir: str) -> dict[str, Any]: session = _require_session() dest_norm = _norm_peer_dest(dest_dir) dest_abs = _safe_local(dest_norm) dest_abs.mkdir(parents=True, exist_ok=True) session.inbound_dest = dest_abs count = 0 for rel in paths: rel = (rel or "").replace("\\", "/").strip() # Keep absolute peer paths (/Users/... or C:/...) for pull RPC. if not (rel.startswith("/") or (len(rel) >= 2 and rel[1] == ":")): rel = rel.lstrip("/") STATE.set_progress(0, 1, rel) session.rpc("pull", path=rel, timeout=30) # Wait until inbound idle after activity deadline = time.monotonic() + 3600 saw = False idle = 0 while time.monotonic() < deadline: if session._inbound.get("fh"): saw = True idle = 0 elif saw: idle += 1 if idle >= 8: # ~0.8s quiet after transfer break time.sleep(0.1) count += 1 session.inbound_dest = STATE.local_root STATE.last_progress = {"done": 1, "total": 1, "label": "완료", "error": None, "busy": False} return {"count": count, "dest_dir": dest_norm} class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def handle(self) -> None: try: super().handle() except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError): # Client closed before the next request line (restart, tab close, aborted fetch). return def log_message(self, fmt: str, *args: Any) -> None: return def _json(self, code: int, payload: dict[str, Any]) -> None: raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") try: self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(raw) except (ConnectionAbortedError, ConnectionResetError, BrokenPipeError, OSError): # Browser navigated away / cancelled long RPC — ignore. return def _read_json(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length") or 0) if length <= 0: return {} return json.loads(self.rfile.read(length).decode("utf-8")) def _read_multipart(self) -> tuple[dict[str, str], list[tuple[str, bytes]]]: ctype = self.headers.get("Content-Type", "") if "multipart/form-data" not in ctype: raise ValueError("multipart required") length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) boundary = None for part in ctype.split(";"): part = part.strip() if part.startswith("boundary="): boundary = part.split("=", 1)[1].strip().strip('"') if not boundary: raise ValueError("no boundary") fields: dict[str, str] = {} files: list[tuple[str, bytes]] = [] for chunk in body.split(b"--" + boundary.encode()): if not chunk or chunk in (b"--\r\n", b"--"): continue if chunk.startswith(b"\r\n"): chunk = chunk[2:] if chunk.endswith(b"\r\n"): chunk = chunk[:-2] if b"\r\n\r\n" not in chunk: continue header_blob, data = chunk.split(b"\r\n\r\n", 1) if data.endswith(b"\r\n"): data = data[:-2] headers = header_blob.decode("utf-8", errors="replace") name = None filename = None for line in headers.split("\r\n"): if line.lower().startswith("content-disposition:"): for token in line.split(";"): token = token.strip() if token.startswith("name="): name = token.split("=", 1)[1].strip().strip('"') if token.startswith("filename="): filename = token.split("=", 1)[1].strip().strip('"') if not name: continue if filename is not None: files.append((filename, data)) else: fields[name] = data.decode("utf-8", errors="replace") return fields, files def do_GET(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) path = parsed.path qs = urllib.parse.parse_qs(parsed.query) if path in ("/", "/index.html"): target = _static_file("index.html") if not target: self._json(500, {"error": "index.html missing"}) return data = target.read_bytes() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) return if path.startswith("/api/"): try: self._api_get(path, qs) except Exception as exc: msg = str(exc).strip() or type(exc).__name__ if isinstance(exc, TimeoutError) or "timed out" in msg.lower(): msg = ( msg if "상대" in msg or "타임아웃" in msg else "상대 응답 시간 초과 — 다시 연결하거나 폴더 새로고침을 시도하세요." ) self._json(400, {"error": msg, "ok": False, "code": type(exc).__name__}) return rel = path.lstrip("/") target = _static_file(rel) if not target: self._json(404, {"error": "not found"}) return data = target.read_bytes() mime = mimetypes.guess_type(str(target))[0] or "application/octet-stream" self.send_response(200) self.send_header("Content-Type", mime) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _api_get(self, path: str, qs: dict[str, list[str]]) -> None: if path == "/api/status": try: devices = list_devices() list_err = None except Exception as exc: # noqa: BLE001 devices = [] list_err = str(exc) ordered = order_devices( devices, serial=STATE.serial, model=STATE.cable_model, ) available = _apply_cable_availability(devices) cable = STATE.active_device or (ordered[0] if ordered else None) handshake = _session_alive() and not STATE.usb_finalized cable_model = None if STATE.link is not None: try: cable_model = STATE.link.model except Exception: pass if not cable_model and STATE.active_device: cable_model = STATE.active_device.model elif not cable_model and cable is not None: cable_model = cable.model if _session_alive() and STATE.usb_finalized and STATE.cable_model: cable_model = STATE.cable_model if devices: if _session_alive() and STATE.usb_finalized: auto = STATE.auto_status elif handshake: auto = "상대 연결 감지 — 자동 동기화 중…" elif STATE.connecting: auto = STATE.auto_status or "상대 프로그램 연결 중…" elif STATE.listening: auto = STATE.auto_status or "케이블 감지됨 — 상대 「연결」 대기 중…" else: auto = "케이블 감지됨 — 「연결」을 누르면 양쪽이 자동 연결됩니다" elif list_err: auto = f"USB 오류: {list_err}" else: auto = ( "케이블 미감지 — python -m juc500_xfer doctor 로 진단하세요" if not STATE.connecting else STATE.auto_status ) self._json( 200, { "connected": _session_alive() and STATE.usb_finalized, "handshake": handshake, "connecting": STATE.connecting, "listening": STATE.listening, "connect_error": STATE.connect_error, "hostname": STATE.peer_name, "host": STATE.peer_name or "", "port": 0, "bind_ip": None, "cable_ip": cable.serial if cable else None, "cable_only": True, "this_host": socket.gethostname(), "local_root": str(STATE.local_root), "local_start": "", "root": STATE.peer_name or "JUC500", "remote_start": "", "remote_receive": "", "auto_status": auto if not _session_alive() else STATE.auto_status, "progress": STATE.last_progress, "device": { "serial": cable.serial if cable else None, "product": cable.product if cable else None, "model": cable_model or (cable.model if cable else None), "count": len(devices), "vid": f"{cable.vid:04x}" if cable else None, "pid": f"{cable.pid:04x}" if cable else None, "if_num": cable.if_num if cable else None, }, "cable_model": cable_model, "selected_cable_model": STATE.cable_model, "available_cable_models": list(available), "active_cable": ( {**_device_dict(cable), "model": cable_model or cable.model} if cable else None ), "devices": [_device_dict(d) for d in ordered], "usb_error": list_err, "wormhole": wormhole_status(), "inputshare": SHARE.status_dict(), "clipboard": _clipboard_status(), **_app_version_info(), }, ) return if path == "/api/peer/ping": if not _session_alive(): self._json(400, {"error": "not connected"}) return try: result = STATE.session.rpc("ping", timeout=8.0) # type: ignore[union-attr] self._json(200, result) except Exception as exc: self._json(500, {"error": str(exc)}) return if path == "/api/doctor": self._json(200, doctor().to_dict()) return if path == "/api/wormhole": self._json(200, wormhole_status()) return if path == "/api/settings/ports": # Reuse settings modal: expose local_root / serial as "ports"-like rows self._json( 200, { "config_path": str(STATE.local_root), "ports": [ { "key": "gui", "label": "GUI 포트", "description": "로컬 웹 UI 포트 (재시작 시 적용)", "protocol": "HTTP", "default": DEFAULT_GUI_PORT, "value": STATE.gui_port, } ], }, ) return if path == "/api/inputshare": self._json(200, SHARE.status_dict()) return if path == "/api/clipboard": self._json(200, _clipboard_status()) return if path == "/api/progress": self._json(200, STATE.last_progress) return if path == "/api/interfaces": devices = [ { "name": d.product, "ip": d.serial, "usb": True, } for d in list_devices() ] self._json(200, {"interfaces": devices}) return if path == "/api/discover": devices = list_devices() peers = [] if devices and _session_alive(): peers.append({"host": STATE.peer_name or "peer", "tcp_port": 0}) self._json(200, {"peers": peers}) return if path == "/api/local/listdir": rel = (qs.get("path") or [""])[0] self._json(200, _list_local_entries(rel)) return if path == "/api/listdir": session = _require_session() rel = (qs.get("path") or [""])[0] result = session.rpc("listdir", path=rel, timeout=45) # Remember last browsed remote folder as paste destination browsed = _norm_peer_dest(str(result.get("path") or rel or "")) if browsed and browsed != ROOTS_PATH: STATE.paste_dest = browsed self._json(200, result) return self._json(404, {"error": f"unknown api {path}"}) def do_POST(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) path = parsed.path try: self._api_post(path) except Exception as e: self._json(400, {"error": str(e), "ok": False, "code": type(e).__name__}) def _api_post(self, path: str) -> None: if path in ("/api/connect", "/api/connect/auto"): try: # Async: HTTP returns immediately; UI polls /api/status. data = start_connect_usb_async() self._json(200, data) except TimeoutError as e: self._json(404, {"ok": False, "error": str(e), "code": "PEER_NOT_FOUND"}) return if path == "/api/disconnect": disconnect_usb() with STATE.lock: STATE.connect_error = None self._json(200, {"ok": True}) return if path == "/api/cable/select": body = self._read_json() model = str(body.get("model") or "").strip().upper() try: data = switch_cable_model(model) self._json(200, data) except ValueError as exc: self._json(400, {"ok": False, "error": str(exc)}) return if path == "/api/quit": self._json(200, {"ok": True, "quitting": True}) request_quit("API quit — 입력공유 끄고 종료") return if path in ("/api/wormhole/kill", "/api/wormhole/stop"): result = kill_wormhole_processes(force=True) STATE.log(result.get("message") or "wormhole kill") self._json(200, result) return if path == "/api/auto/start": start_autolisten() self._json(200, {"ok": True, "listening": STATE.listening}) return if path == "/api/restart/peer": peer_ok = False peer_err: str | None = None if _session_alive() and STATE.session is not None: try: STATE.session.rpc("restart_gui", timeout=12.0) peer_ok = True except Exception as exc: peer_err = str(exc).strip() or type(exc).__name__ else: peer_err = "not connected" self._json( 200 if peer_ok else 400, { "ok": peer_ok, "peer": "scheduled" if peer_ok else "failed", "peer_error": peer_err, }, ) return if path == "/api/restart/both": peer_ok = False peer_err: str | None = None if _session_alive() and STATE.session is not None: try: STATE.session.rpc("restart_gui", timeout=12.0) peer_ok = True except Exception as exc: peer_err = str(exc).strip() or type(exc).__name__ body = { "ok": True, "local": "scheduled", "peer": "scheduled" if peer_ok else "skipped", "peer_error": peer_err, } self._json(200, body) threading.Timer( 0.35, lambda: schedule_restart("양쪽 재시작 — 로컬"), name="juc500-restart-local", ).start() return if path == "/api/mkdir": body = self._read_json() session = _require_session() result = session.rpc("mkdir", path=body.get("path") or "", timeout=15) self._json(200, result) return if path == "/api/local/send": body = self._read_json() result = _send_local_paths(body.get("paths") or [], body.get("dest_dir") or "") self._json(200, result) return if path == "/api/remote/fetch": body = self._read_json() result = _fetch_remote_paths(body.get("paths") or [], body.get("dest_dir") or "") self._json(200, result) return if path == "/api/upload": fields, files = self._read_multipart() dest = _norm_peer_dest(fields.get("dest_dir") or "") session = _require_session() # write temp then send count = 0 with tempfile.TemporaryDirectory() as td: for name, data in files: tmp = Path(td) / Path(name).name tmp.write_bytes(data) remote_name = _join_peer_path(dest, tmp.name) if dest else tmp.name parent = _peer_parent(remote_name) if parent: try: session.rpc("mkdir", path=parent, timeout=15) except Exception: pass session.send_file(tmp, remote_name=remote_name) count += 1 self._json(200, {"count": count, "dest_dir": dest}) return if path == "/api/local/write": fields, files = self._read_multipart() dest = fields.get("dest_dir") or "" dest_abs = _safe_local(_norm_peer_dest(dest)) dest_abs.mkdir(parents=True, exist_ok=True) count = 0 for name, data in files: (dest_abs / Path(name).name).write_bytes(data) count += 1 self._json(200, {"count": count, "dest_dir": dest}) return if path == "/api/settings/ports": body = self._read_json() ports = body.get("ports") or {} if "gui" in ports: STATE.gui_port = int(ports["gui"]) self._json( 200, { "ok": True, "gui_port": STATE.gui_port, "redirect": f"http://127.0.0.1:{STATE.gui_port}/", "message": "포트는 다음 실행 시 적용됩니다 (재시작 없음)", }, ) return if path == "/api/paste-dest": body = self._read_json() raw = _norm_peer_dest(str(body.get("path") or "")) if raw == ROOTS_PATH: raw = "" STATE.paste_dest = raw self._json(200, {"ok": True, "paste_dest": STATE.paste_dest}) return if path == "/api/clipboard/send": info = start_clipboard_send(source="api") code = 200 if info.get("ok") else 400 self._json(code, info) return if path == "/api/inputshare/enable": body = self._read_json() enabled = body.get("enabled") if enabled is None: enabled = body.get("enable") if enabled is None: enabled = not SHARE.enabled if enabled and STATE.session: SHARE.bind_session(STATE.session) SHARE._log = STATE.log info = SHARE.set_enabled(bool(enabled)) self._json(200 if info.get("ok", True) is not False else 400, info) return if path == "/api/inputshare/control": body = self._read_json() target = (body.get("target") or "local").strip().lower() if target not in ("local", "remote"): self._json(400, {"ok": False, "error": "target must be local|remote"}) return if STATE.session: SHARE.bind_session(STATE.session) info = SHARE.set_control(target) code = 200 if info.get("ok", True) else 400 self._json(code, info) return if path == "/api/inputshare/prompt-trust": self._json(200, SHARE.prompt_trust()) return if path == "/api/inputshare/recover": self._json(200, SHARE.recover_local_input()) return if path.startswith("/api/inputshare"): self._json(404, {"ok": False, "error": f"unknown inputshare api {path}"}) return self._json(404, {"error": f"unknown api {path}"}) def _pick_port(host: str, preferred: int) -> int: for port in [preferred, *range(preferred + 1, preferred + 20)]: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: s.bind((host, port)) return port except OSError: continue raise RuntimeError("GUI 포트를 열 수 없습니다") def run_gui(open_browser: bool = True, port: int | None = None, host: str = "127.0.0.1") -> None: global _GUI_SERVER preferred = port or STATE.gui_port or DEFAULT_GUI_PORT listen_port = _pick_port(host, preferred) STATE.gui_port = listen_port # Register early so even abrupt SIGINT/SIGTERM restores KM before exit. atexit.register(_atexit_restore_input) def _on_signal(signum, _frame) -> None: try: restore_all_input(notify_peer=True) except Exception: pass try: disconnect_usb() except Exception: pass # atexit will also run; force exit after restore raise SystemExit(0) for sig in (getattr(signal, "SIGINT", None), getattr(signal, "SIGTERM", None)): if sig is None: continue try: signal.signal(sig, _on_signal) except Exception: pass # Windows: console close / logoff / Ctrl+Break if sys.platform == "win32": try: import ctypes HandlerRoutine = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_uint) def _console_handler(ctrl_type: int) -> bool: # 0=C, 1=Break, 2=Close, 5=Logoff, 6=Shutdown restore_all_input(notify_peer=True) return False # let default terminate continue after restore _handler_ref = HandlerRoutine(_console_handler) ctypes.windll.kernel32.SetConsoleCtrlHandler(_handler_ref, True) # Keep ref so GC does not drop the callback run_gui._win_console_handler = _handler_ref # type: ignore[attr-defined] except Exception: pass # Unstick modifiers/cursor left from a previous force-killed KM session restore_all_input(notify_peer=False) # Cable autorun Wormhole often starts with the volume — clear it on launch. try: result = kill_wormhole_processes(force=True) if result.get("killed"): print(result.get("message")) except Exception as exc: # noqa: BLE001 print(f"Wormhole 자동 종료 건너뜀: {exc}") server = ThreadingHTTPServer((host, listen_port), Handler) _GUI_SERVER = server SHARE.on_escape_quit = request_quit url = f"http://{host}:{listen_port}/" print(f"JUC500 GUI: {url}") print("fileShare와 동일한 듀얼 패널 UI · 케이블 연결 시 양쪽 자동 연결됩니다.") print("입력공유는 연결 시 기본 켜짐 — 한쪽에서 끄면 상대도 자동 동기화됩니다.") print("Esc — 입력공유 끄고 프로그램 종료") start_autolisten() start_autoconnect() start_cable_watch() if open_browser: threading.Timer(0.4, lambda: webbrowser.open(url)).start() try: server.serve_forever() except KeyboardInterrupt: print("\nGUI 종료") finally: _GUI_SERVER = None stop_autolisten() stop_autoconnect() stop_cable_watch() restore_all_input(notify_peer=True) disconnect_usb() try: server.server_close() except Exception: pass