AnabasisChatRemove/services/update_service.py

import json
import os
import re
import urllib.error
import urllib.request
from urllib.parse import urlparse

from PySide6.QtCore import QObject, Signal


def _version_key(version_text):
    parts = [int(x) for x in re.findall(r"\d+", str(version_text))]
    if not parts:
        return (0, 0, 0)
    while len(parts) < 3:
        parts.append(0)
    return tuple(parts[:3])


def _is_newer_version(latest_version, current_version):
    latest_key = _version_key(latest_version)
    current_key = _version_key(current_version)
    return latest_key > current_key


def _sanitize_repo_url(value):
    value = (value or "").strip()
    if not value:
        return ""
    if "://" not in value and value.count("/") == 1:
        return f"https://github.com/{value}"
    parsed = urlparse(value)
    if not parsed.scheme or not parsed.netloc:
        return ""
    clean_path = parsed.path.rstrip("/")
    if clean_path.endswith(".git"):
        clean_path = clean_path[:-4]
    return f"{parsed.scheme}://{parsed.netloc}{clean_path}"


def detect_update_repository_url(configured_url="", configured_repo=""):
    env_url = _sanitize_repo_url(os.getenv("ANABASIS_UPDATE_URL", ""))
    if env_url:
        return env_url
    env_repo = _sanitize_repo_url(os.getenv("ANABASIS_UPDATE_REPOSITORY", ""))
    if env_repo:
        return env_repo
    cfg_url = _sanitize_repo_url(configured_url)
    if cfg_url:
        return cfg_url
    cfg_repo = _sanitize_repo_url(configured_repo)
    if cfg_repo:
        return cfg_repo
    git_config_path = os.path.join(os.path.abspath("."), ".git", "config")
    if not os.path.exists(git_config_path):
        return ""
    try:
        with open(git_config_path, "r", encoding="utf-8") as f:
            content = f.read()
        match = re.search(r"url\s*=\s*((?:https?://|git@)[^\s]+)", content)
        if not match:
            return ""
        remote = match.group(1).strip()
        if remote.startswith("git@"):
            ssh_match = re.match(r"git@([^:]+):(.+?)(?:\.git)?$", remote)
            if ssh_match:
                return _sanitize_repo_url(f"https://{ssh_match.group(1)}/{ssh_match.group(2)}")
        return _sanitize_repo_url(remote)
    except Exception:
        return ""


class UpdateChecker(QObject):
    check_finished = Signal(dict)
    check_failed = Signal(str)

    def __init__(self, repository_url, current_version, request_timeout=8):
        super().__init__()
        self.repository_url = repository_url
        self.current_version = current_version
        self.request_timeout = request_timeout

    def run(self):
        if not self.repository_url:
            self.check_failed.emit("Не задан URL репозитория обновлений.")
            return

        parsed = urlparse(self.repository_url)
        base_url = f"{parsed.scheme}://{parsed.netloc}"
        repo_path = parsed.path.strip("/")
        if not repo_path or repo_path.count("/") < 1:
            self.check_failed.emit("Некорректный URL репозитория обновлений.")
            return

        if parsed.netloc.lower().endswith("github.com"):
            api_url = f"https://api.github.com/repos/{repo_path}/releases/latest"
        else:
            api_url = f"{base_url}/api/v1/repos/{repo_path}/releases/latest"
        releases_url = f"{base_url}/{repo_path}/releases"
        request = urllib.request.Request(
            api_url,
            headers={
                "Accept": "application/vnd.github+json",
                "User-Agent": "AnabasisManager-Updater",
            },
        )
        try:
            with urllib.request.urlopen(request, timeout=self.request_timeout) as response:
                release_data = json.loads(response.read().decode("utf-8"))
        except urllib.error.HTTPError as e:
            self.check_failed.emit(f"Ошибка HTTP при проверке обновлений: {e.code}")
            return
        except urllib.error.URLError as e:
            self.check_failed.emit(f"Сетевая ошибка при проверке обновлений: {e}")
            return
        except Exception as e:
            self.check_failed.emit(f"Не удалось проверить обновления: {e}")
            return

        latest_tag = release_data.get("tag_name") or release_data.get("name") or ""
        latest_version = latest_tag.lstrip("vV").strip()
        html_url = release_data.get("html_url") or releases_url
        assets = release_data.get("assets") or []
        download_url = ""
        download_name = ""
        checksum_url = ""
        for asset in assets:
            url = asset.get("browser_download_url", "")
            if url.lower().endswith(".zip"):
                download_url = url
                download_name = asset.get("name", "")
                break
        if not download_url and assets:
            download_url = assets[0].get("browser_download_url", "")
            download_name = assets[0].get("name", "")

        for asset in assets:
            name = asset.get("name", "").lower()
            if not name:
                continue
            is_checksum_asset = name.endswith(".sha256") or name.endswith(".sha256.txt") or name in ("checksums.txt", "sha256sums.txt")
            if not is_checksum_asset:
                continue
            if download_name and (download_name.lower() in name or name in (f"{download_name.lower()}.sha256", f"{download_name.lower()}.sha256.txt")):
                checksum_url = asset.get("browser_download_url", "")
                break
            if not checksum_url:
                checksum_url = asset.get("browser_download_url", "")

        self.check_finished.emit(
            {
                "repository_url": self.repository_url,
                "latest_version": latest_version,
                "current_version": self.current_version,
                "latest_tag": latest_tag,
                "release_url": html_url,
                "download_url": download_url,
                "download_name": download_name,
                "checksum_url": checksum_url,
                "has_update": _is_newer_version(latest_version, self.current_version),
            }
        )