""" Минимальная проверка новой версии через GitHub Releases API (без сторонних зависимостей). Ограничение частоты запросов: не чаще одного раза в час на машину (кэш в каталоге данных приложения). Поддерживается If-None-Match (ETag) для ответа 304. """ from __future__ import annotations import json import os import sys import time from itertools import zip_longest from pathlib import Path from typing import Any, Dict, Optional, Tuple from urllib.error import HTTPError, URLError from urllib.request import Request, urlopen REPO = "Flowseal/tg-ws-proxy" RELEASES_LATEST_API = f"https://api.github.com/repos/{REPO}/releases/latest" RELEASES_PAGE_URL = f"https://github.com/{REPO}/releases/latest" # Не чаще одного полного запроса к API в час (без учёта 304 с тем же ETag). _MIN_FETCH_INTERVAL_SEC = 3600.0 _state: Dict[str, Any] = { "checked": False, "has_update": False, "ahead_of_release": False, "latest": None, "html_url": None, "error": None, } def _cache_file() -> Optional[Path]: try: if sys.platform == "win32": root = Path(os.environ.get("APPDATA", str(Path.home()))) / "TgWsProxy" elif sys.platform == "darwin": root = Path.home() / "Library/Application Support/TgWsProxy" else: xdg = os.environ.get("XDG_CONFIG_HOME") root = (Path(xdg).expanduser() if xdg else Path.home() / ".config") / "TgWsProxy" root.mkdir(parents=True, exist_ok=True) return root / ".update_check_cache.json" except OSError: return None def _load_cache(path: Optional[Path]) -> Dict[str, Any]: if not path or not path.is_file(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} def _save_cache(path: Optional[Path], data: Dict[str, Any]) -> None: if not path: return try: path.write_text(json.dumps(data), encoding="utf-8") except OSError: pass def _parse_version_tuple(s: str) -> tuple: s = (s or "").strip().lstrip("vV") if not s: return (0,) parts = [] for seg in s.split("."): digits = "".join(c for c in seg if c.isdigit()) if digits: try: parts.append(int(digits)) except ValueError: parts.append(0) else: parts.append(0) return tuple(parts) if parts else (0,) def _version_gt(a: str, b: str) -> bool: """True, если версия a новее b (простое сравнение по сегментам).""" ta = _parse_version_tuple(a) tb = _parse_version_tuple(b) for x, y in zip_longest(ta, tb, fillvalue=0): if x > y: return True if x < y: return False return False def _apply_release_tag( tag: str, html_url: str, current_version: str, ) -> None: global _state if not tag: _state["has_update"] = False _state["ahead_of_release"] = False _state["latest"] = None _state["html_url"] = html_url.strip() or RELEASES_PAGE_URL return latest_clean = tag.lstrip("vV") cur = (current_version or "").strip().lstrip("vV") _state["latest"] = latest_clean _state["html_url"] = html_url.strip() or RELEASES_PAGE_URL _state["has_update"] = _version_gt(latest_clean, cur) _state["ahead_of_release"] = bool(latest_clean) and _version_gt( cur, latest_clean ) def fetch_latest_release( timeout: float = 12.0, etag: Optional[str] = None, ) -> Tuple[Optional[dict], Optional[str], int]: """ GET releases/latest. Возвращает (data или None при 304, etag или None, HTTP-код). """ headers = { "Accept": "application/vnd.github+json", "User-Agent": "tg-ws-proxy-update-check", } if etag: headers["If-None-Match"] = etag req = Request( RELEASES_LATEST_API, headers=headers, method="GET", ) try: with urlopen(req, timeout=timeout) as resp: code = getattr(resp, "status", None) or resp.getcode() new_etag = resp.headers.get("ETag") raw = resp.read().decode("utf-8", errors="replace") return json.loads(raw), new_etag, int(code) except HTTPError as e: if e.code == 304: hdrs = e.headers new_etag = hdrs.get("ETag") if hdrs else None return None, new_etag or etag, 304 raise def run_check(current_version: str) -> None: """Запрашивает последний релиз и обновляет внутреннее состояние.""" global _state _state["checked"] = True _state["error"] = None cache_path = _cache_file() cache = _load_cache(cache_path) now = time.time() last_attempt = float(cache.get("last_attempt_at") or 0) if last_attempt and (now - last_attempt) < _MIN_FETCH_INTERVAL_SEC: tag = (cache.get("tag_name") or "").strip() if tag: _apply_release_tag(tag, cache.get("html_url") or "", current_version) return err = cache.get("last_error") _state["error"] = ( err if err else "Проверка обновлений отложена (интервал между запросами)." ) _state["has_update"] = False _state["ahead_of_release"] = False _state["latest"] = None _state["html_url"] = RELEASES_PAGE_URL return etag = (cache.get("etag") or "").strip() or None try: data, new_etag, code = fetch_latest_release(etag=etag) cache["last_attempt_at"] = now if code == 304: tag = (cache.get("tag_name") or "").strip() url = (cache.get("html_url") or "").strip() or RELEASES_PAGE_URL _apply_release_tag(tag, url, current_version) if new_etag: cache["etag"] = new_etag _save_cache(cache_path, cache) return assert data is not None tag = (data.get("tag_name") or "").strip() html_url = (data.get("html_url") or "").strip() or RELEASES_PAGE_URL if not tag: _state["has_update"] = False _state["ahead_of_release"] = False _state["latest"] = None _state["html_url"] = html_url else: _apply_release_tag(tag, html_url, current_version) if new_etag: cache["etag"] = new_etag cache["tag_name"] = tag cache["html_url"] = html_url cache.pop("last_error", None) _save_cache(cache_path, cache) except (HTTPError, URLError, OSError, TimeoutError, ValueError, json.JSONDecodeError) as e: cache["last_attempt_at"] = now msg = str(e) if isinstance(e, HTTPError) and e.code == 403: msg = ( "GitHub API вернул 403 (лимит или доступ). Повторите позже." ) cache["last_error"] = msg _save_cache(cache_path, cache) _state["error"] = msg _state["has_update"] = False _state["ahead_of_release"] = False _state["latest"] = None _state["html_url"] = RELEASES_PAGE_URL def get_status() -> Dict[str, Any]: """Снимок состояния после run_check (для подписей в настройках).""" return dict(_state)