"""Command-line interface for JUC500 file transfer.""" from __future__ import annotations import argparse import os import sys import time from pathlib import Path from . import __version__ from .protocol import Session from .usb_link import list_devices, Juc500Link, doctor from .wormhole_kill import kill_wormhole_processes, wormhole_status def _progress(prefix: str): last = [0.0] def cb(done: int, total: int) -> None: now = time.monotonic() if now - last[0] < 0.1 and done < total: return last[0] = now pct = (done / total * 100) if total else 100.0 mb = done / (1024 * 1024) tot = total / (1024 * 1024) print(f"\r{prefix} {mb:.2f}/{tot:.2f} MiB ({pct:5.1f}%)", end="", flush=True) if done >= total: print() return cb def cmd_info(_args: argparse.Namespace) -> int: devices = list_devices() if not devices: print("연결된 JUC500 장치가 없습니다. 진단: python -m juc500_xfer doctor") return 1 for d in devices: print( f"{d.vid:04x}:{d.pid:04x} bus={d.bus} addr={d.address} " f"serial={d.serial!r} product={d.product!r}" ) return 0 def cmd_doctor(_args: argparse.Namespace) -> int: rep = doctor() print(f"libusb: {rep.libusb_path or '(없음)'}") if rep.libusb_error: print(f"libusb 오류: {rep.libusb_error}") print(f"USB 장치 수(libusb): {rep.usb_device_count}") print(f"JUC500 후보: {len(rep.juc500)}") for j in rep.juc500: print(f" - {j}") if rep.sample_devices: print("USB 샘플:") for line in rep.sample_devices: print(f" {line}") if rep.pnp_vid0711: print("Windows PnP VID_0711:") for row in rep.pnp_vid0711: print(f" - {row}") wh = wormhole_status() print(f"Wormhole 프로세스: {wh['count']} — {wh['message']}") for p in wh.get("processes") or []: print(f" - {p.get('name')} pid={p.get('pid')}") print("안내:") for h in rep.hints: print(f" * {h}") return 0 if rep.ok else 1 def cmd_kill_wormhole(_args: argparse.Namespace) -> int: result = kill_wormhole_processes(force=True) print(result.get("message") or "") if result.get("killed"): for p in result["killed"]: print(f" 종료: {p.get('name')} pid={p.get('pid')}") if result.get("remaining"): for p in result["remaining"]: print(f" 잔여: {p.get('name')} pid={p.get('pid')}") return 1 return 0 def cmd_probe(args: argparse.Namespace) -> int: with Juc500Link(serial=args.serial) as link: print(f"열림: serial={link.serial!r}") link.drain(0.3) payload = b"JUC500-PROBE-" + os.urandom(8) n = link.write_data(payload) print(f"DATA OUT 기록: {n} bytes") # If Windows peer echoes or our own framed app is listening, we may see traffic for i in range(10): data = link.read_data(timeout_ms=300) if data: print(f"DATA IN[{i}]: {data[:64]!r} ({len(data)} bytes)") ctrl = link.read_ctrl(timeout_ms=50) if ctrl: print(f"CTRL IN[{i}]: {ctrl[:64]!r}") print("probe 완료 (상대 앱이 없으면 IN 데이터는 없을 수 있음)") return 0 def cmd_send(args: argparse.Namespace) -> int: path = Path(args.file) with Juc500Link(serial=args.serial) as link: session = Session(link, peer_name=args.name, on_log=print) session.start() try: peer = session.wait_peer(timeout=args.timeout) print(f"연결됨: {peer}") session.send_file(path, progress=_progress("전송")) finally: session.stop() return 0 def cmd_recv(args: argparse.Namespace) -> int: dest = Path(args.dest) with Juc500Link(serial=args.serial) as link: session = Session(link, peer_name=args.name, on_log=print) session.start() try: peer = session.wait_peer(timeout=args.timeout) print(f"연결됨: {peer}") def allow(name: str, size: int) -> bool: if args.yes: print(f"수신 수락: {name} ({size} bytes)") return True ans = input(f"수신할까요? {name} ({size} bytes) [y/N] ").strip().lower() return ans in ("y", "yes", "ㅛ") session.receive_file(dest, allow=allow, progress=_progress("수신"), timeout=args.recv_timeout) finally: session.stop() return 0 def cmd_gui(args: argparse.Namespace) -> int: from .gui_server import run_gui run_gui(open_browser=not args.no_browser, port=args.port) return 0 def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="juc500", description="Apple Silicon Mac ↔ Windows 11 JUC500(Smart Data Link) 파일 전송", ) p.add_argument("--version", action="version", version=f"juc500-xfer {__version__}") sub = p.add_subparsers(dest="cmd", required=True) s = sub.add_parser("info", help="연결된 JUC500 장치 목록") s.set_defaults(func=cmd_info) s = sub.add_parser("doctor", help="USB/libusb/WinUSB 진단") s.set_defaults(func=cmd_doctor) s = sub.add_parser("kill-wormhole", help="공식 Wormhole/GoBridge 프로세스 종료") s.set_defaults(func=cmd_kill_wormhole) s = sub.add_parser("probe", help="USB 인터페이스 점유 및 송수신 프로브") s.add_argument("--serial", default=None) s.set_defaults(func=cmd_probe) s = sub.add_parser("send", help="파일 전송") s.add_argument("file") s.add_argument("--serial", default=None) s.add_argument("--name", default=None, help="이 쪽 peer 이름") s.add_argument("--timeout", type=float, default=60.0, help="상대 연결 대기(초)") s.set_defaults(func=cmd_send) s = sub.add_parser("recv", help="파일 수신") s.add_argument("--dest", default=".", help="저장 폴더") s.add_argument("--serial", default=None) s.add_argument("--name", default=None) s.add_argument("--timeout", type=float, default=60.0) s.add_argument("--recv-timeout", type=float, default=3600.0) s.add_argument("-y", "--yes", action="store_true", help="수락 확인 생략") s.set_defaults(func=cmd_recv) s = sub.add_parser("gui", help="fileShare 스타일 웹 GUI 실행") s.add_argument("--port", type=int, default=None, help="GUI 포트 (기본 8765)") s.add_argument("--no-browser", action="store_true", help="브라우저 자동 실행 안 함") s.set_defaults(func=cmd_gui) return p def main(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) if getattr(args, "name", None) is None and args.cmd in ("send", "recv"): host = os.uname().nodename if hasattr(os, "uname") else "host" args.name = f"{host}" try: return args.func(args) except KeyboardInterrupt: print("\n중단됨") return 130 except Exception as exc: print(f"오류: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())