286 lines
10 KiB
Python
286 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
업데이트 서비스 모듈
|
|
애플리케이션 업데이트를 확인하고 설치합니다.
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import requests
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import urllib3
|
|
|
|
# SSL 경고 숨기기
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
from PySide6.QtCore import QObject, QTimer, QThread, Signal
|
|
|
|
from core.config import ConfigManager
|
|
from core.signals import GlobalSignals
|
|
from core.constants import APP_VERSION, UPDATE_CHECK_INTERVAL
|
|
from core.logger import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Supabase Configuration
|
|
SUPABASE_URL = "https://kong.humetrain.me"
|
|
SUPABASE_KEY = "INSERT_ANON_KEY_HERE" # TODO: Replace with actual key or load from secure config
|
|
|
|
|
|
class UpdateChecker(QObject):
|
|
"""업데이트 확인 워커"""
|
|
|
|
update_available = Signal(str, str) # 새 버전, 다운로드 URL
|
|
no_update = Signal()
|
|
error = Signal(str)
|
|
|
|
def __init__(self, current_version: str):
|
|
super().__init__()
|
|
self.current_version = current_version
|
|
|
|
def run(self):
|
|
"""업데이트 확인"""
|
|
try:
|
|
# Supabase에서 최신 버전 정보 조회
|
|
# 테이블: app_versions
|
|
# 쿼리: select=version,download_url&order=created_at.desc&limit=1
|
|
|
|
headers = {
|
|
"apikey": SUPABASE_KEY,
|
|
"Authorization": f"Bearer {SUPABASE_KEY}"
|
|
}
|
|
|
|
url = f"{SUPABASE_URL}/rest/v1/app_versions?select=version,download_url&order=created_at.desc&limit=1"
|
|
|
|
# SSL 인증서 오류 방지를 위해 verify=False 추가 (임시)
|
|
# 실제 배포 시에는 올바른 인증서 처리가 필요함
|
|
try:
|
|
response = requests.get(url, headers=headers, timeout=10, verify=False)
|
|
except requests.exceptions.SSLError:
|
|
logger.warning("SSL 인증 실패. 업데이트 확인을 건너뜁니다.")
|
|
self.no_update.emit()
|
|
return
|
|
except requests.exceptions.ConnectionError:
|
|
logger.warning("서버 연결 실패. 업데이트 확인을 건너뜁니다.")
|
|
self.no_update.emit()
|
|
return
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if data and len(data) > 0:
|
|
latest_info = data[0]
|
|
latest_version = latest_info.get("version")
|
|
download_url = latest_info.get("download_url")
|
|
|
|
if latest_version and self._compare_versions(latest_version, self.current_version) > 0:
|
|
self.update_available.emit(latest_version, download_url)
|
|
return
|
|
elif response.status_code == 401:
|
|
# 키가 없거나 잘못된 경우 (개발 중)
|
|
logger.warning("Supabase 인증 실패. 업데이트 확인을 건너뜁니다.")
|
|
elif response.status_code == 525:
|
|
# SSL Handshake Failed
|
|
logger.warning("SSL 핸드셰이크 실패 (Cloudflare 525). 업데이트 확인을 건너뜁니다.")
|
|
else:
|
|
logger.warning(f"업데이트 확인 실패: {response.status_code} {response.text}")
|
|
|
|
self.no_update.emit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"업데이트 확인 중 오류 발생: {e}")
|
|
# 백그라운드 확인 시 사용자에게 에러를 띄우지 않도록 함
|
|
# self.error.emit(str(e))
|
|
self.no_update.emit()
|
|
|
|
@staticmethod
|
|
def _compare_versions(v1: str, v2: str) -> int:
|
|
"""
|
|
버전 비교
|
|
Returns: v1 > v2: 1, v1 == v2: 0, v1 < v2: -1
|
|
"""
|
|
def parse_version(v):
|
|
return [int(x) for x in v.split('.') if x.isdigit()]
|
|
|
|
parts1 = parse_version(v1)
|
|
parts2 = parse_version(v2)
|
|
|
|
for i in range(max(len(parts1), len(parts2))):
|
|
p1 = parts1[i] if i < len(parts1) else 0
|
|
p2 = parts2[i] if i < len(parts2) else 0
|
|
|
|
if p1 > p2:
|
|
return 1
|
|
elif p1 < p2:
|
|
return -1
|
|
|
|
return 0
|
|
|
|
|
|
class UpdateService(QObject):
|
|
"""
|
|
업데이트 서비스 클래스
|
|
주기적으로 업데이트를 확인하고 알림을 표시합니다.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self.config = ConfigManager()
|
|
self.signals = GlobalSignals()
|
|
|
|
self._timer = QTimer()
|
|
self._timer.timeout.connect(self.check_for_updates)
|
|
|
|
self._thread: Optional[QThread] = None
|
|
self._checker: Optional[UpdateChecker] = None
|
|
|
|
self._show_no_update = False
|
|
|
|
logger.info("업데이트 서비스 초기화 완료")
|
|
|
|
def start(self):
|
|
"""서비스 시작"""
|
|
if not self.config.get('app', 'check_updates', True):
|
|
return
|
|
|
|
# 주기적 확인 시작
|
|
interval = self.config.get('app', 'update_check_interval', UPDATE_CHECK_INTERVAL)
|
|
self._timer.start(interval * 1000)
|
|
|
|
logger.info("업데이트 서비스 시작")
|
|
|
|
def stop(self):
|
|
"""서비스 중지"""
|
|
self._timer.stop()
|
|
|
|
if self._thread and self._thread.isRunning():
|
|
self._thread.quit()
|
|
self._thread.wait()
|
|
|
|
logger.info("업데이트 서비스 중지")
|
|
|
|
def check_for_updates(self, show_no_update: bool = False):
|
|
"""
|
|
업데이트 확인
|
|
Args:
|
|
show_no_update: 업데이트 없을 때도 알림 표시
|
|
"""
|
|
self._show_no_update = show_no_update
|
|
|
|
if self._thread and self._thread.isRunning():
|
|
return
|
|
|
|
self._thread = QThread()
|
|
self._checker = UpdateChecker(APP_VERSION)
|
|
self._checker.moveToThread(self._thread)
|
|
|
|
self._thread.started.connect(self._checker.run)
|
|
self._checker.update_available.connect(self._on_update_available)
|
|
self._checker.no_update.connect(self._on_no_update)
|
|
self._checker.error.connect(self._on_error)
|
|
self._checker.update_available.connect(self._thread.quit)
|
|
self._checker.no_update.connect(self._thread.quit)
|
|
self._checker.error.connect(self._thread.quit)
|
|
|
|
self._thread.start()
|
|
|
|
logger.debug("업데이트 확인 시작")
|
|
|
|
def _on_update_available(self, version: str, download_url: str):
|
|
"""업데이트 가능"""
|
|
logger.info(f"업데이트 가능: v{version}")
|
|
|
|
# UI에 알림 (다운로드 URL도 함께 전달하기 위해 시그널 변경 필요할 수 있음)
|
|
# 현재는 GlobalSignals.update_available(str) 만 정의되어 있다고 가정
|
|
# 따라서 메인 윈도우에서 이 서비스를 직접 참조하거나, 시그널을 확장해야 함.
|
|
# 여기서는 편의상 GlobalSignals를 통해 버전만 알리고,
|
|
# 실제 업데이트 진행 시 이 서비스의 trigger_update를 호출하도록 함.
|
|
|
|
# 임시 저장 (메인 윈도우에서 접근 가능하도록)
|
|
self.latest_version = version
|
|
self.download_url = download_url
|
|
|
|
self.signals.update_available.emit(version)
|
|
|
|
# 시스템 알림
|
|
self.signals.notification.emit(
|
|
"업데이트 가능",
|
|
f"새 버전 v{version}이 있습니다.",
|
|
"info"
|
|
)
|
|
|
|
def _on_no_update(self):
|
|
"""업데이트 없음"""
|
|
logger.debug("최신 버전입니다.")
|
|
if self._show_no_update:
|
|
self.signals.status_message.emit("최신 버전입니다.", 3000)
|
|
|
|
def _on_error(self, error_msg: str):
|
|
"""업데이트 확인 오류"""
|
|
if self._show_no_update:
|
|
self.signals.status_message.emit(f"업데이트 확인 실패: {error_msg}", 3000)
|
|
|
|
def trigger_update(self):
|
|
"""
|
|
업데이트 프로세스 시작
|
|
1. updater.exe를 임시 폴더로 복사
|
|
2. updater.exe 실행
|
|
3. 앱 종료
|
|
"""
|
|
if not hasattr(self, 'download_url') or not self.download_url:
|
|
logger.error("다운로드 URL이 없습니다.")
|
|
return
|
|
|
|
try:
|
|
# 현재 실행 파일 경로
|
|
if getattr(sys, 'frozen', False):
|
|
base_dir = Path(sys.executable).parent
|
|
exe_name = Path(sys.executable).name
|
|
else:
|
|
# 개발 환경
|
|
base_dir = Path(__file__).parent.parent
|
|
exe_name = "main.py" # 개발 환경에서는 재시작이 다를 수 있음
|
|
|
|
updater_src = base_dir / "updater.exe"
|
|
|
|
# 개발 환경 대응: updater.exe가 없으면 updater.py 사용 (복잡하므로 생략, 배포 환경 기준)
|
|
if not updater_src.exists():
|
|
logger.warning("updater.exe를 찾을 수 없습니다. (개발 환경일 수 있음)")
|
|
# 개발 환경에서는 단순히 로그만 남기고 종료
|
|
return
|
|
|
|
# 임시 폴더 생성
|
|
temp_dir = Path(os.environ.get('TEMP', '')) / "handover_updater"
|
|
if temp_dir.exists():
|
|
shutil.rmtree(temp_dir)
|
|
temp_dir.mkdir(parents=True)
|
|
|
|
# updater 복사
|
|
updater_dst = temp_dir / "updater.exe"
|
|
shutil.copy2(updater_src, updater_dst)
|
|
|
|
# updater 실행
|
|
# args: --url [URL] --target [DIR] --restart [EXE_NAME]
|
|
cmd = [
|
|
str(updater_dst),
|
|
"--url", self.download_url,
|
|
"--target", str(base_dir),
|
|
"--restart", exe_name
|
|
]
|
|
|
|
logger.info(f"업데이트 시작: {cmd}")
|
|
subprocess.Popen(cmd)
|
|
|
|
# 앱 종료 요청
|
|
self.signals.app_quit_requested.emit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"업데이트 트리거 실패: {e}")
|
|
self.signals.status_message.emit(f"업데이트 시작 실패: {e}", 3000)
|
|
|
|
|
|
|