live-app-translator/packaging/windows_smoke.py

#!/usr/bin/env python3
"""Windows 전용 경로 점검.

리눅스 개발 환경에서는 확인이 불가능한 것들만 모아놨다. 여기서 실패하면
"리눅스에서 테스트 다 통과했는데 윈도우에서 안 되더라" 가 되는 지점이다.

    python packaging/windows_smoke.py                  # 소스에서 점검
    python packaging/windows_smoke.py --exe dist\\LiveSub.exe   # 빌드된 exe 점검

결과는 artifacts/ 에 로그와 스크린샷으로 남는다. CI 가 그걸 업로드하므로
원격에서도 실제 화면을 눈으로 확인할 수 있다.
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
ARTIFACTS = ROOT / "artifacts"
sys.path.insert(0, str(ROOT / "src"))

# Qt 플랫폼 플러그인은 QApplication 을 만들기 *전에* 정해야 한다.
# 표시할 화면이 없는데 그냥 두면 플러그인 로드 실패로 SIGABRT 가 나면서
# 아무 결과도 못 남기고 죽는다. 점검기가 그렇게 죽으면 쓸모가 없다.
if sys.platform != "win32" and not (
    os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")
):
    os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")

_results: list[tuple[str, bool, str]] = []


def _qt_app():
    """QApplication 을 한 번만 만들어 재사용한다."""
    from PySide6.QtWidgets import QApplication

    return QApplication.instance() or QApplication([])


def check(name: str):
    """점검 하나. 예외가 나도 나머지는 계속 돈다."""

    def wrapper(fn):
        try:
            detail = fn() or ""
            ok = True
        except Exception as exc:  # noqa: BLE001 - 점검 결과로 기록하는 게 목적
            detail, ok = f"{type(exc).__name__}: {exc}", False
        _results.append((name, ok, str(detail)))
        print(f"  [{'OK  ' if ok else '실패'}] {name}  {detail}")
        return fn

    return wrapper


# --- 소스 기준 점검 ---------------------------------------------------------


def run_source_checks() -> None:
    print(f"\n=== Windows 전용 경로 (실행 플랫폼: {sys.platform}) ===")
    if sys.platform != "win32":
        print("  주의: Windows 가 아니라 Windows 전용 항목은 당연히 실패합니다.")

    @check("WASAPI 루프백 장치 열거")
    def _devices():
        from livesub.audio.wasapi_loopback import WasapiLoopbackCapture

        if not WasapiLoopbackCapture.available():
            raise RuntimeError("PyAudioWPatch 없음 — 오디오 캡처 불가")
        sources = WasapiLoopbackCapture.list_sources()
        if not sources:
            raise RuntimeError("루프백 장치가 하나도 없음")
        return f"{len(sources)}개: {sources[0].label}"

    @check("소리 내는 프로그램 열거 (pycaw)")
    def _processes():
        from livesub.audio.process_loopback import ProcessLoopbackCapture

        sources = ProcessLoopbackCapture.list_sources()
        return f"{len(sources)}개" + (f" (예: {sources[0].label})" if sources else " (지금 소리 내는 앱 없음)")

    @check("프로그램별 캡처 보조 프로그램")
    def _helper():
        from livesub.audio.process_loopback import helper_path

        path = helper_path()
        if path is None:
            raise RuntimeError("livesub_capture.exe 없음 — 장치 전체 캡처로 폴백됨")
        return str(path)

    @check("전역 단축키 등록/해제")
    def _hotkey():
        from livesub.ui.hotkeys import GlobalHotkeys

        _qt_app()
        keys = GlobalHotkeys()
        if not keys.supported:
            raise RuntimeError("이 플랫폼은 전역 단축키 미지원")
        fired = []
        if not keys.register("Ctrl+Alt+F12", lambda: fired.append(1)):
            raise RuntimeError(keys.last_error)
        keys.unregister_all()
        return "Ctrl+Alt+F12 등록·해제 성공"

    @check("모니터 열거")
    def _screens():
        from livesub.ui.placement import list_screens

        _qt_app()
        screens = list_screens()
        if not screens:
            raise RuntimeError("모니터를 찾지 못함")
        return " / ".join(s.label for s in screens)

    @check("GPU 인식")
    def _gpu():
        from livesub.models import detect_gpu

        gpu = detect_gpu()
        if not gpu.available:
            raise RuntimeError(gpu.reason)
        return f"{gpu.name} · {gpu.total_vram_mb / 1024:.1f}GB"

    @check("사용 가능한 모델 티어")
    def _tiers():
        from livesub.models.manager import has_torch, tier_availability
        from livesub.models.tiers import ordered_tiers

        usable = [t.name for t in ordered_tiers() if tier_availability(t)[0]]
        return f"torch={'있음' if has_torch() else '없음'} → {', '.join(usable)}"

    @check("자막 오버레이 실제 렌더링")
    def _overlay():
        from livesub.config import AppConfig
        from livesub.ui.main_window import MainWindow

        app = _qt_app()
        cfg = AppConfig()
        cfg.models.preload_on_start = False
        win = MainWindow(cfg)
        try:
            win.show()
            win.overlay.set_subtitle_visible(True)
            win.overlay.set_line("적이 왼쪽에서 들어온다, 지금 바로 빠져!", "Enemy from the left")
            for _ in range(10):
                app.processEvents()
                time.sleep(0.05)
            ARTIFACTS.mkdir(exist_ok=True)
            win.grab().save(str(ARTIFACTS / "windows-main.png"))
            win.overlay.grab().save(str(ARTIFACTS / "windows-overlay.png"))
        finally:
            win.engine.shutdown()
            win.overlay.close()
            win.close()
        return "스크린샷 2장 저장 (artifacts/)"


# --- 전 파이프라인 점검 -----------------------------------------------------


def run_pipeline_check(timeout_s: int = 900) -> None:
    """오디오 파일 하나를 넣어 한국어 자막이 나오는지 끝까지 본다.

    개별 부품이 다 통과해도 이어 붙이면 안 되는 경우가 있다. 이 점검이
    통과하면 "음성인식 -> 번역 -> 자막" 이 실제로 동작한다는 뜻이다.

    GPU 없이 CPU 로 돌린다. 서버의 GPU 없는 Windows VM 에서도 그대로 된다.
    모델은 처음 한 번만 받고 그 뒤로는 캐시를 쓴다.
    """
    print(f"\n=== 전 파이프라인 (오디오 -> 자막), 플랫폼: {sys.platform} ===")

    @check("오디오 파일 -> 한국어 자막")
    def _pipeline():
        from livesub.config import AppConfig
        from livesub.core.engine import TranslationEngine

        clip = ROOT / "tests" / "fixtures" / "en_callout_16k.wav"
        if not clip.is_file():
            raise RuntimeError(f"테스트 음성이 없습니다: {clip}")

        cfg = AppConfig()
        cfg.models.tier = "lightning"   # CI 에서는 가장 가벼운 티어로 충분하다
        cfg.models.device = "cpu"       # GPU 없는 VM 에서도 돌아야 한다
        cfg.models.preload_on_start = True
        cfg.audio.backend = "file"
        cfg.audio.target_process_name = str(clip)
        cfg.audio.silence_ms = 400
        cfg.performance.game_mode = False

        lines: list = []
        errors: list[str] = []

        def on_status(status) -> None:
            if status.state.value == "error":
                errors.append(status.message)

        engine = TranslationEngine(cfg, on_line=lines.append, on_status=on_status)
        engine.start()
        deadline = time.time() + timeout_s
        try:
            while time.time() < deadline and not errors:
                if any(line.is_final for line in lines):
                    break
                time.sleep(0.5)
        finally:
            engine.shutdown()

        if errors:
            raise RuntimeError(f"엔진 오류: {errors[0]}")
        finals = [line for line in lines if line.is_final]
        if not finals:
            raise RuntimeError(f"{timeout_s}초 안에 자막이 나오지 않았습니다")

        line = finals[0]
        if "enemy" not in line.source_text.lower():
            raise RuntimeError(f"음성인식이 빗나갔습니다: {line.source_text!r}")
        if not any("\uac00" <= ch <= "\ud7a3" for ch in line.translated_text):
            raise RuntimeError(f"한국어 자막이 아닙니다: {line.translated_text!r}")

        ARTIFACTS.mkdir(exist_ok=True)
        (ARTIFACTS / "pipeline.txt").write_text(
            f"원문: {line.source_text}\n자막: {line.translated_text}\n",
            encoding="utf-8",
        )
        return f"{line.source_text!r} -> {line.translated_text!r}"


# --- GPU 전용 점검 ----------------------------------------------------------


def run_gpu_checks() -> None:
    """GPU 가 있는 머신에서만 의미가 있는 것들.

    GPU 는 한 VM 만 독점하므로(컨슈머 NVIDIA 는 vGPU 미지원) GPU 없는 CI VM
    에서는 확인할 수 없다. 그래서 별도 러너 라벨(windows-gpu)로 떼어놨다.
    여기서는 GPU 가 없으면 **분명히 실패**해야 한다 — GPU 를 확인하겠다고
    부른 작업이 조용히 통과하면 안 된다.
    """
    print(f"\n=== GPU 전용 (플랫폼: {sys.platform}) ===")

    @check("CUDA 사용 가능")
    def _cuda():
        import torch

        if not torch.cuda.is_available():
            raise RuntimeError(
                "torch 가 CUDA 를 못 씁니다. CPU 빌드가 깔렸는지 확인하세요 "
                "(pip install torch --index-url https://download.pytorch.org/whl/cu128)"
            )
        idx = torch.cuda.current_device()
        free, total = torch.cuda.mem_get_info(idx)
        return f"{torch.cuda.get_device_name(idx)} · {total / 1024**3:.1f}GB (여유 {free / 1024**3:.1f}GB)"

    @check("앱의 GPU 인식")
    def _detect():
        from livesub.models import detect_gpu

        gpu = detect_gpu()
        if not gpu.available:
            raise RuntimeError(gpu.reason)
        return f"{gpu.name} · {gpu.total_vram_mb / 1024:.1f}GB"

    @check("VRAM 상한이 실제로 걸리는지")
    def _vram():
        from livesub.models.manager import apply_vram_limit

        if not apply_vram_limit(0.35):
            raise RuntimeError("상한 적용 실패 — 게임과 같이 쓸 때 VRAM 을 못 막습니다")
        return "35% 상한 적용됨"

    @check("티어 게이트가 뭐라고 하는지")
    def _gate():
        """앱이 각 torch 티어를 쓸 수 있다고 하는지, 아니면 왜 막는지.

        여기서는 판정하지 않고 그대로 보고만 한다. VRAM 이 작은 GPU 에서
        4·5티어가 막히는 건 정상이라 실패로 처리하면 안 된다. 진짜 판정은
        아래 "실제 로드" 가 한다 — 게이트가 된다고 한 걸 못 올리면 그게 버그다.
        """
        from livesub.models.manager import detect_gpu, has_torch, tier_availability
        from livesub.models.tiers import MTBackend, ordered_tiers

        if not has_torch():
            raise RuntimeError("torch 없음 — GPU 검증에는 일반 설치가 필요합니다")
        gpu = detect_gpu()
        parts = []
        for t in ordered_tiers():
            if t.mt.backend is not MTBackend.TRANSFORMERS:
                continue
            ok, reason = tier_availability(t, gpu)
            parts.append(f"{t.name}={'가능' if ok else reason}")
        return " / ".join(parts)

    @check("게이트가 된다고 한 티어가 실제로 올라가는지")
    def _real_load():
        """게이트 판정을 **실물로 검증**한다.

        게이트가 "가능"이라 한 torch 티어를 실제로 올려 번역까지 시킨다.
        여기서 죽으면 앱이 사용자에게 거짓말을 하고 있다는 뜻이다.

        게이트가 전부 막았다면 검증할 대상이 없다. 그건 이 GPU 의 한계이지
        회귀가 아니므로 실패로 처리하지 않고 이유를 남긴다.
        """
        from livesub.models.glossary import Glossary
        from livesub.models.manager import detect_gpu, tier_availability
        from livesub.models.tiers import MTBackend, ordered_tiers
        from livesub.models.translator import create_translator

        gpu = detect_gpu()
        allowed, blocked = [], []
        for t in ordered_tiers():
            if t.mt.backend is not MTBackend.TRANSFORMERS:
                continue
            ok, reason = tier_availability(t, gpu)
            (allowed if ok else blocked).append(t if ok else f"{t.name}({reason})")

        if not allowed:
            return (
                "검증 대상 없음 — 이 GPU 로는 torch 티어를 못 씁니다. "
                f"{'; '.join(blocked)}"
            )

        tier = allowed[-1]   # 열린 것 중 가장 무거운 것
        translator = create_translator(tier.mt, device="cuda")
        translator.load()
        try:
            out = translator.translate("Enemy coming from the left.", "en", "ko", Glossary())
        finally:
            translator.unload()
        if not any("\uac00" <= ch <= "\ud7a3" for ch in out):
            raise RuntimeError(f"{tier.name}: 한국어가 아닙니다 — {out!r}")
        return f"{tier.name} 로드·번역 성공 {out!r}"

    @check("GPU 로 실제 번역 (1~3티어 CUDA 경로)")
    def _translate():
        from livesub.models.glossary import Glossary
        from livesub.models.packs import build_glossary
        from livesub.models.tiers import get_tier
        from livesub.models.translator import create_translator

        tier = get_tier("swift")   # CT2 지만 device=cuda 로 GPU 경로를 탄다
        translator = create_translator(tier.mt, device="cuda")
        translator.load()
        glossary = build_glossary(Glossary(), ["fps-common", "pubg"])
        try:
            out = translator.translate("Enemy coming from the left.", "en", "ko", glossary)
        finally:
            translator.unload()
        if not any("\uac00" <= ch <= "\ud7a3" for ch in out):
            raise RuntimeError(f"한국어가 아닙니다: {out!r}")
        ARTIFACTS.mkdir(exist_ok=True)
        (ARTIFACTS / "gpu-translate.txt").write_text(out + "\n", encoding="utf-8")
        return repr(out)


# --- 빌드된 exe 점검 --------------------------------------------------------


def run_exe_check(exe: Path) -> None:
    print(f"\n=== 빌드된 exe 점검: {exe} ===")

    @check("exe 존재와 크기")
    def _size():
        if not exe.is_file():
            raise RuntimeError("파일 없음")
        mb = exe.stat().st_size / 1024 / 1024
        if mb > 1500:
            raise RuntimeError(f"{mb:.0f}MB — 너무 큼. torch 가 섞여 들어갔는지 확인")
        return f"{mb:.1f} MB"

    @check("exe 실행 후 8초 생존")
    def _launch():
        # 창을 띄우고 잠깐 살아 있는지 본다. 바로 죽으면 의존성 누락이다.
        proc = subprocess.Popen([str(exe)], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        time.sleep(8)
        alive = proc.poll() is None
        if not alive:
            err = (proc.stderr.read() or b"").decode("utf-8", "replace")[-800:]
            raise RuntimeError(f"즉시 종료 (코드 {proc.returncode}): {err}")
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()
        return "정상 기동"


def main() -> int:
    parser = argparse.ArgumentParser(description="Windows 전용 점검")
    parser.add_argument("--exe", type=Path, help="빌드된 exe 를 점검")
    parser.add_argument(
        "--pipeline", action="store_true",
        help="오디오 파일로 전 파이프라인 점검 (모델을 내려받으므로 느립니다)",
    )
    parser.add_argument(
        "--gpu", action="store_true",
        help="GPU 가 있는 머신에서만 되는 항목 점검 (CUDA·VRAM 상한·4/5티어)",
    )
    args = parser.parse_args()

    if args.exe:
        run_exe_check(args.exe)
    elif args.gpu:
        run_gpu_checks()
    elif args.pipeline:
        run_pipeline_check()
    else:
        run_source_checks()

    ARTIFACTS.mkdir(exist_ok=True)
    report = "\n".join(
        f"[{'OK' if ok else 'FAIL'}] {name}: {detail}" for name, ok, detail in _results
    )
    (ARTIFACTS / "windows-smoke.txt").write_text(report, encoding="utf-8")

    failed = [n for n, ok, _ in _results if not ok]
    print(f"\n{len(_results) - len(failed)}/{len(_results)} 통과")
    if failed:
        print("실패:", ", ".join(failed))
    # 오디오 장치·보조 프로그램·GPU 는 러너 환경에 따라 없을 수 있어 경고로만 둔다.
    # 다만 --gpu 로 부른 경우엔 GPU 가 없는 것 자체가 실패다.
    lenient = {
        "프로그램별 캡처 보조 프로그램",
        "소리 내는 프로그램 열거 (pycaw)",
        "WASAPI 루프백 장치 열거",
        "GPU 인식",
    }
    fatal = failed if args.gpu else [n for n in failed if n not in lenient]
    return 1 if fatal else 0


if __name__ == "__main__":
    raise SystemExit(main())