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


# --- 빌드된 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="오디오 파일로 전 파이프라인 점검 (모델을 내려받으므로 느립니다)",
    )
    args = parser.parse_args()

    if args.exe:
        run_exe_check(args.exe)
    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))
    # 오디오 장치·보조 프로그램은 러너 환경에 따라 없을 수 있어 경고로만 둔다.
    fatal = [n for n in failed if n not in {
        "프로그램별 캡처 보조 프로그램",
        "소리 내는 프로그램 열거 (pycaw)",
        "WASAPI 루프백 장치 열거",
        "GPU 인식",
    }]
    return 1 if fatal else 0


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