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)


# 업데이트 체크 결과 상수 (info=None 일 때 reason 식별)
UP_TO_DATE = 'up_to_date'
NETWORK_ERROR = 'network_error'
NO_RELEASE = 'no_release'      # latest 응답 자체가 없음 (404 등)
NO_ASSET = 'no_asset'          # release는 있으나 main.exe 자산 없음


def check_for_update(current_version: str, timeout: int = 5):
    """Releases API 조회.

    Returns:
        (ReleaseInfo, None) — 새 버전 있음
        (None, UP_TO_DATE) — 이미 최신
        (None, NETWORK_ERROR) — 네트워크/타임아웃 실패
        (None, NO_RELEASE) — 저장소 비공개/리소스 없음 (404)
        (None, NO_ASSET) — 새 버전이지만 main.exe 자산 미첨부
    """
    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.HTTPError as e:
        if e.code == 404:
            return None, NO_RELEASE
        return None, NETWORK_ERROR
    except (urllib.error.URLError, json.JSONDecodeError, TimeoutError, OSError):
        return None, NETWORK_ERROR

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

    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, NO_ASSET

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


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:
        # CREATE_NO_WINDOW + DETACHED_PROCESS — updater.exe도 windowed 빌드라
        # 정상적으로는 콘솔이 안 뜨지만, 안전하게 두 플래그 모두 적용해서
        # 어떤 환경에서도 cmd 창 깜빡임이 보이지 않도록.
        if sys.platform == 'win32':
            DETACHED_PROCESS = 0x00000008
            CREATE_NO_WINDOW = 0x08000000
            creationflags = DETACHED_PROCESS | CREATE_NO_WINDOW
        else:
            creationflags = 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 찾기.

    1) 메인 .exe와 같은 폴더 (정상 배포 상태)
    2) PyInstaller _MEIPASS — main.exe에 내장된 fallback
       (main.py 시작 시 _ensure_updater_extracted()가 #1로 복사하므로 평소엔 미사용)
    """
    candidate = _exe_dir() / 'updater.exe'
    if candidate.exists():
        return candidate

    # _MEIPASS fallback — main.py 시작 추출이 권한 부족 등으로 실패한 경우
    if getattr(sys, 'frozen', False):
        meipass = getattr(sys, '_MEIPASS', None)
        if meipass:
            bundled = Path(meipass) / 'updater.exe'
            if bundled.exists():
                # 사용자 TEMP 폴더에 복사 (메인 종료 후에도 살아남도록)
                import tempfile, shutil
                tmp = Path(tempfile.gettempdir()) / 'clockout_updater.exe'
                try:
                    shutil.copy2(str(bundled), str(tmp))
                    return tmp
                except OSError:
                    pass

    return None