Clock_out_Time_Calculator/utils/updater_client.py

"""
업데이트 클라이언트 — Gitea/GitHub Releases 호환.

자체 호스팅 Gitea 인스턴스 사용. Gitea API가 GitHub과 호환되어 endpoint URL만
바꾸면 동일 로직 동작.

흐름:
1. `check_for_update()`: Releases API → 최신 태그 조회 → 현재 버전과 비교
2. `download_update(asset_url)`: 새 .exe를 임시 폴더에 다운로드
3. `apply_update(new_exe)`: updater.exe 실행 + 메인 앱 종료

환경변수로 URL 오버라이드 가능 (테스트/마이그레이션 용도):
- CLOCKOUT_RELEASES_API: 전체 API endpoint URL
- CLOCKOUT_ASSET_NAME: 다운로드할 자산 파일명 (기본 main.exe)
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import urllib.request
import urllib.error
from dataclasses import dataclass
from pathlib import Path
from typing import Optional

# 자체 호스팅 Gitea 인스턴스 (GitHub 호환 API)
GIT_HOST = 'https://kindnick-git.duckdns.org'
GIT_OWNER = 'kindnick'
GIT_REPO = 'Clock_out_Time_Calculator'

# Gitea API endpoint: /api/v1/repos/{owner}/{repo}/releases/latest
# (GitHub의 경우: https://api.github.com/repos/{owner}/{repo}/releases/latest)
DEFAULT_RELEASES_API = f'{GIT_HOST}/api/v1/repos/{GIT_OWNER}/{GIT_REPO}/releases/latest'
RELEASES_API = os.environ.get('CLOCKOUT_RELEASES_API', DEFAULT_RELEASES_API)

# 다운로드할 .exe 자산 이름 (Releases 첨부 파일명과 일치해야 함)
ASSET_NAME = os.environ.get('CLOCKOUT_ASSET_NAME', 'main.exe')

USER_AGENT = 'ClockOutCalculator-Updater/1.0'


@dataclass
class ReleaseInfo:
    version: str  # 'v2.1.0' or '2.1.0'
    asset_url: str  # main.exe 다운로드 URL
    notes: str  # 릴리스 노트 (markdown)
    published_at: str

    @property
    def version_clean(self) -> str:
        """'v2.1.0' → '2.1.0'"""
        return self.version.lstrip('vV')


def _parse_version(s: str) -> tuple:
    """semver-ish 파싱. 'v2.1.0' / '2.1' → (2,1,0). 비교용."""
    s = s.lstrip('vV').strip()
    parts = s.split('.')
    out = []
    for p in parts:
        # 'rc1' 등 접미사 제거
        digits = ''
        for c in p:
            if c.isdigit():
                digits += c
            else:
                break
        out.append(int(digits) if digits else 0)
    while len(out) < 3:
        out.append(0)
    return tuple(out[:3])


def is_newer(remote: str, local: str) -> bool:
    """remote가 local보다 최신이면 True."""
    return _parse_version(remote) > _parse_version(local)


def check_for_update(current_version: str, timeout: int = 5) -> Optional[ReleaseInfo]:
    """GitHub Releases API 조회. 새 버전 있으면 ReleaseInfo, 없으면 None.

    네트워크 오류 시 None 반환 (앱 시작을 막지 않음).
    """
    try:
        req = urllib.request.Request(RELEASES_API, headers={'User-Agent': USER_AGENT})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            data = json.loads(resp.read())
    except (urllib.error.URLError, json.JSONDecodeError, TimeoutError):
        return None

    tag = data.get('tag_name', '')
    if not tag or not is_newer(tag, current_version):
        return None

    asset_url = None
    for asset in data.get('assets', []):
        if asset.get('name') == ASSET_NAME:
            asset_url = asset.get('browser_download_url')
            break

    if not asset_url:
        return None

    return ReleaseInfo(
        version=tag,
        asset_url=asset_url,
        notes=data.get('body', ''),
        published_at=data.get('published_at', ''),
    )


def download_update(asset_url: str, dest_dir: Optional[Path] = None,
                     progress_cb=None) -> Optional[Path]:
    """새 .exe를 임시 폴더에 다운로드.

    Args:
        asset_url: GitHub Releases asset URL
        dest_dir: 저장 위치 (기본: 메인 .exe 옆에 main_new.exe)
        progress_cb: callable(downloaded_bytes, total_bytes) — 진행률 콜백

    Returns:
        다운로드된 파일 경로, 실패 시 None.
    """
    if dest_dir is None:
        dest_dir = _exe_dir()
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    dest = dest_dir / 'main_new.exe'
    if dest.exists():
        try:
            dest.unlink()
        except OSError:
            return None

    try:
        req = urllib.request.Request(asset_url, headers={'User-Agent': USER_AGENT})
        with urllib.request.urlopen(req, timeout=30) as resp:
            total = int(resp.headers.get('Content-Length', 0))
            downloaded = 0
            chunk_size = 64 * 1024
            with open(dest, 'wb') as f:
                while True:
                    chunk = resp.read(chunk_size)
                    if not chunk:
                        break
                    f.write(chunk)
                    downloaded += len(chunk)
                    if progress_cb:
                        progress_cb(downloaded, total)
        return dest
    except (urllib.error.URLError, OSError, TimeoutError):
        if dest.exists():
            try:
                dest.unlink()
            except OSError:
                pass
        return None


def apply_update(new_exe: Path) -> bool:
    """updater.exe를 실행하여 파일 교체 + 재시작 트리거.

    호출 직후 메인 앱은 종료되어야 함 (Qt: QApplication.quit()).
    """
    target_exe = _current_exe()
    updater_exe = _find_updater()
    if not updater_exe or not target_exe:
        return False

    pid = os.getpid()
    try:
        DETACHED_PROCESS = 0x00000008 if sys.platform == 'win32' else 0
        creationflags = DETACHED_PROCESS if sys.platform == 'win32' else 0
        subprocess.Popen(
            [str(updater_exe),
             '--pid', str(pid),
             '--new', str(new_exe),
             '--target', str(target_exe)],
            creationflags=creationflags,
            close_fds=True,
        )
        return True
    except OSError:
        return False


def _exe_dir() -> Path:
    """현재 .exe가 있는 폴더 (PyInstaller frozen) 또는 main.py 위치."""
    if getattr(sys, 'frozen', False):
        return Path(sys.executable).parent
    return Path(__file__).resolve().parent.parent


def _current_exe() -> Optional[Path]:
    """현재 실행 중인 메인 .exe. 개발 환경(.py)에선 None."""
    if getattr(sys, 'frozen', False):
        return Path(sys.executable)
    return None


def _find_updater() -> Optional[Path]:
    """updater.exe 찾기. 메인 .exe와 같은 폴더에 있어야 함."""
    candidate = _exe_dir() / 'updater.exe'
    if candidate.exists():
        return candidate
    return None