juc500/dist-offline/juc500-offline-update-20260721-124606/juc500_xfer/cli.py

"""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}")
    if rep.ioreg_vid0711:
        print("macOS IORegistry VID_0711:")
        for row in rep.ioreg_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())