juc500/juc500_xfer/gui_server.py

"""fileShare 스타일 로컬 웹 GUI — JUC500 USB bulk 브릿지."""

from __future__ import annotations

import json
import mimetypes
import os
import socket
import tempfile
import threading
import time
import urllib.parse
import webbrowser
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Optional

from .protocol import Session
from .usb_link import Juc500Link, list_devices, doctor

DEFAULT_GUI_PORT = 8765


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
        self.gui_port = DEFAULT_GUI_PORT
        self.auto_status = "JUC500 케이블을 연결하세요"
        self.last_progress = {"done": 0, "total": 0, "label": "", "error": None, "busy": False}
        self.connecting = False
        self.log_lines: list[str] = []

    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()


def _session_alive() -> bool:
    return STATE.session is not None and STATE.session._connected.is_set()


def _safe_local(rel: str) -> Path:
    rel = (rel or "").replace("\\", "/").lstrip("/")
    target = (STATE.local_root / rel).resolve()
    try:
        target.relative_to(STATE.local_root)
    except ValueError as exc:
        raise PermissionError("경로가 로컬 루트를 벗어납니다") from exc
    return target


def _list_local_entries(rel: str) -> dict[str, Any]:
    path = _safe_local(rel)
    if not path.exists():
        raise FileNotFoundError(rel or "/")
    if not path.is_dir():
        raise NotADirectoryError(rel)
    entries = []
    for child in path.iterdir():
        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": rel.replace("\\", "/").strip("/"), "entries": entries}


def connect_usb(timeout: float = 45.0) -> dict[str, Any]:
    with STATE.lock:
        if _session_alive():
            return {
                "ok": True,
                "host": STATE.peer_name or "peer",
                "hostname": STATE.peer_name or "peer",
                "port": 0,
                "start_path": "",
                "receive_path": "",
            }
        if STATE.connecting:
            raise RuntimeError("이미 연결 중입니다")
        STATE.connecting = True
        STATE.auto_status = "USB 링크 여는 중…"

    try:
        devices = list_devices()
        if not devices:
            raise RuntimeError("JUC500(Smart Data Link) 장치가 없습니다. 케이블을 연결하세요.")
        link = Juc500Link(serial=STATE.serial)
        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.start()
        STATE.auto_status = "상대 프로그램 대기 중…"
        peer = session.wait_peer(timeout=timeout)
        with STATE.lock:
            STATE.link = link
            STATE.session = session
            STATE.peer_name = peer
            STATE.auto_status = f"연결됨 · {peer}"
            STATE.connecting = False
        return {
            "ok": True,
            "host": peer,
            "hostname": peer,
            "port": 0,
            "start_path": "",
            "receive_path": "",
        }
    except Exception:
        with STATE.lock:
            STATE.connecting = False
            if STATE.session:
                try:
                    STATE.session.stop()
                except Exception:
                    pass
            if STATE.link:
                try:
                    STATE.link.close()
                except Exception:
                    pass
            STATE.session = None
            STATE.link = None
            STATE.auto_status = "연결 실패 — 다시 시도하세요"
        raise


def disconnect_usb() -> None:
    with STATE.lock:
        session = STATE.session
        link = STATE.link
        STATE.session = None
        STATE.link = None
        STATE.peer_name = None
        STATE.auto_status = "연결 해제됨"
        STATE.connecting = False
    if session:
        try:
            session.stop()
        except Exception:
            pass
    if link:
        try:
            link.close()
        except Exception:
            pass


def _require_session() -> Session:
    if not STATE.session or not STATE.session._connected.is_set():
        raise RuntimeError("연결되지 않았습니다")
    return STATE.session


def _send_local_paths(paths: list[str], dest_dir: str) -> dict[str, Any]:
    session = _require_session()
    dest = (dest_dir or "").replace("\\", "/").strip("/")
    count = 0

    def ensure_remote_dir(rel_dir: str) -> None:
        rel_dir = (rel_dir or "").replace("\\", "/").strip("/")
        if not rel_dir or rel_dir == ".":
            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 = f"{dest}/{local.name}".strip("/") if dest else local.name
            ensure_remote_dir(str(Path(remote_name).parent.as_posix()))
            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)
            session.send_file(local, remote_name=remote_name)
            count += 1
        elif local.is_dir():
            base = f"{dest}/{local.name}".strip("/") 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 = f"{base}/{rel_under}".strip("/")
                ensure_remote_dir(str(Path(remote_name).parent.as_posix()))
                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_rel = (dest_dir or "").replace("\\", "/").strip("/")
    dest_abs = _safe_local(dest_rel)
    dest_abs.mkdir(parents=True, exist_ok=True)
    session.inbound_dest = dest_abs
    count = 0
    for rel in paths:
        rel = rel.replace("\\", "/").lstrip("/")
        # Ask peer what it is — listdir parent and find, or just pull
        # Pull: peer pushes file(s). For directories, peer walks.
        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_rel}


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    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")
        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)

    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:
                self._json(400, {"error": str(exc)})
            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)
            cable = devices[0] if devices else None
            if devices:
                auto = STATE.auto_status if _session_alive() else "케이블 감지됨 — 연결을 누르세요"
            elif list_err:
                auto = f"USB 오류: {list_err}"
            else:
                auto = (
                    "JUC500 미감지 — python -m juc500_xfer doctor 로 진단하세요"
                    if not STATE.connecting
                    else STATE.auto_status
                )
            self._json(
                200,
                {
                    "connected": _session_alive(),
                    "connecting": STATE.connecting,
                    "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,
                        "count": len(devices),
                        "vid": f"{cable.vid:04x}" if cable else None,
                        "pid": f"{cable.pid:04x}" if cable else None,
                    },
                    "usb_error": list_err,
                    "inputshare": {
                        "supported": False,
                        "enabled": False,
                        "trusted": False,
                        "status": "JUC500 빌드에서는 입력공유 미지원",
                    },
                },
            )
            return

        if path == "/api/doctor":
            self._json(200, doctor().to_dict())
            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,
                {
                    "supported": False,
                    "enabled": False,
                    "trusted": False,
                    "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=20)
            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:
                data = connect_usb()
                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()
            self._json(200, {"ok": True})
            return

        if path == "/api/auto/start":
            self._json(200, {"ok": True})
            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 = 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 = f"{dest}/{tmp.name}".strip("/") if dest else tmp.name
                    parent = Path(remote_name).parent.as_posix()
                    if parent and 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(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.startswith("/api/inputshare"):
            self._json(200, {"supported": False, "ok": False, "error": "입력공유 미지원"})
            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:
    preferred = port or STATE.gui_port or DEFAULT_GUI_PORT
    listen_port = _pick_port(host, preferred)
    STATE.gui_port = listen_port
    server = ThreadingHTTPServer((host, listen_port), Handler)
    url = f"http://{host}:{listen_port}/"
    print(f"JUC500 GUI: {url}")
    print("fileShare와 동일한 듀얼 패널 UI · 케이블 연결 후 「연결」을 누르세요.")
    if open_browser:
        threading.Timer(0.4, lambda: webbrowser.open(url)).start()
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nGUI 종료")
    finally:
        disconnect_usb()
        server.server_close()