lintunes/lintunes/updater.py

"""Self-update via git.

LinTunes runs straight from its git checkout (editable install), so an update
is just a fast-forward `git pull` followed by an app relaunch. The Updater
periodically fetches the checkout's upstream and reports how many commits
behind we are; `update_async()` applies them with `--ff-only` (never a merge,
never a conflicted tree). Network git calls run on daemon threads — same
pattern as lastfm.py — and report back through Qt signals.

When the package isn't running from a git checkout (or has no upstream), the
updater disables itself silently and the version display is just a label.
"""

import subprocess
import threading
from pathlib import Path

from PyQt6.QtCore import QObject, QTimer, pyqtSignal

FIRST_CHECK_MS = 10 * 1000              # shortly after launch, off the startup path
RECHECK_INTERVAL_MS = 4 * 3600 * 1000   # then every 4 hours


class Updater(QObject):
    update_available = pyqtSignal(int)   # commits behind upstream
    update_applied = pyqtSignal()        # pull succeeded; caller restarts the app
    update_failed = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)
        self._root = str(Path(__file__).resolve().parents[1])
        self._busy = False
        self._timer = None
        self._enabled = None  # probed lazily — see enabled

    @property
    def enabled(self) -> bool:
        """Whether this checkout is a git repo with an upstream.

        Probed on first use rather than in __init__: the two `git rev-parse`
        calls have a 15 s timeout each, and running them during startup put
        that stall on the main thread before the window was up. First use is
        now `_check`, on a background thread.
        """
        if self._enabled is None:
            self._enabled = (
                self._git("rev-parse", "--show-toplevel") is not None
                and self._git("rev-parse", "--abbrev-ref", "@{upstream}")
                is not None)
        return self._enabled

    def short_hash(self) -> str | None:
        """The checkout's current commit, for telling builds apart."""
        return self._git("rev-parse", "--short", "HEAD")

    def start_checking(self):
        """First check shortly after launch, then every few hours."""
        QTimer.singleShot(FIRST_CHECK_MS, self.check_async)
        self._timer = QTimer(self)
        self._timer.setInterval(RECHECK_INTERVAL_MS)
        self._timer.timeout.connect(self.check_async)
        self._timer.start()

    def check_async(self):
        if self._busy:
            return
        self._busy = True
        threading.Thread(target=self._check, daemon=True).start()

    def update_async(self):
        if not self.enabled or self._busy:
            return
        self._busy = True
        threading.Thread(target=self._update, daemon=True).start()

    # ---- internals (run on daemon threads) ----

    def _check(self):
        try:
            if not self.enabled:
                return
            if self._git("fetch", "--quiet", timeout=60) is None:
                return  # offline / host unreachable — perfectly normal, stay quiet
            behind = self._git("rev-list", "--count", "HEAD..@{upstream}")
            if behind and behind.isdigit() and int(behind) > 0:
                self.update_available.emit(int(behind))
        finally:
            self._busy = False

    def _update(self):
        try:
            proc = subprocess.run(
                ["git", "-C", self._root, "pull", "--ff-only", "--quiet"],
                capture_output=True, text=True, timeout=120)
            if proc.returncode == 0:
                self.update_applied.emit()
            else:
                lines = [l.strip() for l in
                         (proc.stderr or proc.stdout).strip().splitlines()
                         if l.strip()]
                # git's actual reason is the error:/fatal: line, not the
                # trailing "Aborting"
                message = next(
                    (l for l in lines if l.startswith(("error:", "fatal:"))),
                    lines[-1] if lines else "git pull failed")
                self.update_failed.emit(message)
        except Exception as e:
            self.update_failed.emit(str(e))
        finally:
            self._busy = False

    def _git(self, *args, timeout=15) -> str | None:
        """Run git in the repo root; stripped stdout, or None on any failure."""
        try:
            proc = subprocess.run(["git", "-C", self._root, *args],
                                  capture_output=True, text=True,
                                  timeout=timeout)
        except Exception:
            return None
        if proc.returncode != 0:
            return None
        return proc.stdout.strip()