rigdoctor/src/rigdoctor/core/updates.py

"""Update check (M13): ask the Gitea releases API for the latest version + notes.

Stdlib-only (urllib). The Gitea instance requires sign-in, so updates are gated to
account holders via a Personal Access Token (D18): set $RIGDOCTOR_TOKEN or save one
with `rigdoctor login`. Returns the latest tag, its release notes (body), and a clear
state for the UI; `apply_update` performs the no-root self-update.
"""

from __future__ import annotations

import functools
import json
import shutil
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

from .. import __version__
from ..config import load_token

GITEA_BASE = "https://git.jesseyvanofferen.com"
REPO = "jessey/rigdoctor"
LATEST_API = f"{GITEA_BASE}/api/v1/repos/{REPO}/releases/latest"
RELEASES_PAGE = f"{GITEA_BASE}/{REPO}/releases"
TOKEN_PAGE = f"{GITEA_BASE}/user/settings/applications"

# Update states
NO_TOKEN = "no-token"
AUTH = "auth"
NETWORK = "network"
UP_TO_DATE = "up-to-date"
AVAILABLE = "available"


APT_PACKAGE = "rigdoctor"


def _dpkg_owns(path: Path) -> bool:
    """True if dpkg reports `path` belongs to a package (i.e. an apt/.deb install)."""
    if not shutil.which("dpkg"):
        return False
    try:
        r = subprocess.run(["dpkg", "-S", str(path)], capture_output=True, text=True, timeout=5)
    except (subprocess.SubprocessError, OSError):
        return False
    return r.returncode == 0 and APT_PACKAGE in r.stdout


@functools.lru_cache(maxsize=1)
def install_kind() -> str:
    """How RigDoctor was installed: 'apt' (.deb), 'pip' (venv/.run), or 'dev' (source checkout).

    Decides which updater to use: only 'pip' can self-update in place; apt is root/dpkg-managed
    and source is VCS-managed, so those are guided rather than auto-applied.
    """
    pkg = Path(__file__).resolve().parents[1]  # .../rigdoctor
    if _dpkg_owns(pkg / "__init__.py"):
        return "apt"
    if sys.prefix != sys.base_prefix:  # inside a venv → the pip/.run install
        return "pip"
    if (pkg.parents[1] / "pyproject.toml").exists():  # repo checkout
        return "dev"
    if str(pkg).startswith("/usr/") or "/dist-packages/" in str(pkg):
        return "apt"  # system-managed but no dpkg record — still don't pip
    return "pip"


def update_hint(kind: str | None = None) -> str:
    """Human guidance for installs that can't self-update via pip (apt / source)."""
    kind = kind or install_kind()
    if kind == "apt":
        return ("Installed via apt — update with:\n"
                f"  sudo apt update && sudo apt install --only-upgrade {APT_PACKAGE}")
    if kind == "dev":
        return "Running from a source checkout — update with `git pull`."
    return ""


def _parse(version: str) -> tuple[int, ...]:
    return tuple(int(p) for p in version.lstrip("vV").split(".") if p.isdigit())


def is_newer(latest: str, current: str = __version__) -> bool:
    try:
        return _parse(latest) > _parse(current)
    except (ValueError, AttributeError):
        return False


def fetch_latest(timeout: float = 5.0) -> tuple[str | None, str, str | None]:
    """Return (tag, notes, error). error is NO_TOKEN/AUTH/NETWORK, or None on success."""
    token = load_token()
    if not token:
        return (None, "", NO_TOKEN)
    req = urllib.request.Request(
        LATEST_API,
        headers={"Accept": "application/json", "Authorization": f"token {token}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310 (https)
            data = json.load(resp)
        return (data.get("tag_name") or None, (data.get("body") or "").strip(), None)
    except urllib.error.HTTPError as exc:
        return (None, "", AUTH if exc.code in (401, 403) else NETWORK)
    except Exception:
        return (None, "", NETWORK)


def check_latest(timeout: float = 5.0) -> str | None:
    """Convenience: latest tag or None (ignores notes/error)."""
    tag, _notes, _error = fetch_latest(timeout)
    return tag


def update_state(timeout: float = 5.0) -> tuple[str, str | None, str]:
    """Return (state, tag, notes). state in NO_TOKEN/AUTH/NETWORK/UP_TO_DATE/AVAILABLE."""
    tag, notes, error = fetch_latest(timeout)
    if error:
        return (error, None, "")
    if tag and is_newer(tag):
        return (AVAILABLE, tag, notes)
    return (UP_TO_DATE, tag, notes)


def list_releases(limit: int = 15, timeout: float = 6.0) -> tuple[list[tuple[str, str, str]], str | None]:
    """Return ([(tag, date, notes), …], error) for the in-app changelog."""
    token = load_token()
    if not token:
        return ([], NO_TOKEN)
    req = urllib.request.Request(
        f"{GITEA_BASE}/api/v1/repos/{REPO}/releases?limit={limit}",
        headers={"Accept": "application/json", "Authorization": f"token {token}"},
    )
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310 (https)
            data = json.load(resp)
        return ([
            (r.get("tag_name") or "?", (r.get("published_at") or "")[:10], (r.get("body") or "").strip())
            for r in data
        ], None)
    except urllib.error.HTTPError as exc:
        return ([], AUTH if exc.code in (401, 403) else NETWORK)
    except Exception:
        return ([], NETWORK)


def apply_update(tag: str) -> tuple[int, str]:
    """Update to `tag` using the method matching how RigDoctor was installed.

    Only pip/venv installs are upgraded in place (authenticated pip install of
    `rigdoctor[gui] @ git+https://oauth2:<token>@…/rigdoctor.git@<tag>`). apt and source
    installs can't be (root/dpkg- or VCS-managed), so they return guidance instead of
    attempting pip. Returns (exit_code, output) with the token scrubbed.
    """
    kind = install_kind()
    if kind != "pip":
        return (1, update_hint(kind))
    token = load_token()
    if not token:
        return (1, "No update token configured. Run `rigdoctor login`.")
    host = GITEA_BASE.split("://", 1)[1]
    ref = f"rigdoctor[gui] @ git+https://oauth2:{token}@{host}/{REPO}.git@{tag}"
    cmd = [sys.executable, "-m", "pip", "install", "--upgrade", ref]
    try:
        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
        out = (proc.stdout + proc.stderr).replace(token, "***")
        return (proc.returncode, out)
    except (subprocess.SubprocessError, OSError) as exc:
        return (1, str(exc).replace(token, "***"))