zapret-kvn/xray_fluent/updates/update_signal.py

"""Связка слушателя сервера с окном программы.

Слушатель (``release_watch.ReleaseWatcher``) живёт в фоновом потоке и про Qt
не знает. Здесь его разрешения превращаются в сигнал Qt, а окно получает
простые ответы: ведёт ли сервер очередь, разрешена ли версия, занят ли
человек и что сказать серверу после обновления.
"""

from __future__ import annotations

import threading
import time
from collections.abc import Callable

from PyQt6.QtCore import QObject, QTimer, pyqtSignal

from ..constants import APP_VERSION
from .auto_update import WHY_RE
from .release_watch import (
    ACTIVITY_FULLSCREEN,
    ACTIVITY_TRAY,
    ACTIVITY_WINDOW,
    ReleaseWatcher,
    fullscreen_app_active,
    screen_state,
)

# Как часто окно сообщает слушателю, видно ли оно и включено ли подключение.
# Слушатель читает это из своего потока и Qt не трогает.
PRESENCE_REFRESH_MS = 15 * 1000
# Сколько действует разрешение сервера: в этот срок программа повторяет
# сорвавшуюся установку сама (второго разрешения сервер не пришлёт).
GRANT_LIFETIME_S = 60 * 60


class UpdateSignal(QObject):
    """Разрешения сервера на обновление — для окна программы."""

    # Сервер разрешил обновиться до этой версии.
    granted = pyqtSignal(str)
    # Версия вышла, но очередь на скачивание до программы ещё не дошла.
    queued = pyqtSignal(str)

    def __init__(
        self,
        *,
        is_enabled: Callable[[], bool],
        window_shown: Callable[[], bool],
        connected: Callable[[], bool],
        parent: QObject | None = None,
        watcher_factory: Callable[..., ReleaseWatcher] = ReleaseWatcher,
    ) -> None:
        super().__init__(parent)
        self._window_shown = window_shown
        self._connected = connected
        self._lock = threading.Lock()
        self._presence: dict[str, bool] = {}
        self._granted_at: dict[str, float] = {}
        # Версии, которые надо поставить сразу: установленную сервер отозвал.
        self._urgent: set[str] = set()
        self._report: dict[str, str] = {}
        self._watcher = watcher_factory(
            current_version=APP_VERSION,
            on_release=self._on_release,
            on_queued=lambda version: self.queued.emit(str(version)),
            on_revoked=self._on_revoked,
            is_enabled=is_enabled,
            activity=self._activity,
            pending_report=self._pending_report,
            report_delivered=self._report_delivered,
        )
        self._timer = QTimer(self)
        self._timer.setInterval(PRESENCE_REFRESH_MS)
        self._timer.timeout.connect(self.refresh_presence)

    def start(self) -> None:
        from zapret_update import screen

        self.refresh_presence()
        self._timer.start()
        self._watcher.start()
        # Память об игре: экран проверяется и между вопросами серверу.
        screen.start_sampler()

    def stop(self) -> None:
        self._timer.stop()
        self._watcher.stop()

    # --- для окна (поток интерфейса) ---

    def refresh_presence(self) -> None:
        """Запоминает, видно ли окно и включено ли подключение."""
        try:
            presence = {"window": bool(self._window_shown()), "connected": bool(self._connected())}
        except Exception:
            return
        with self._lock:
            self._presence = presence

    def queue_reachable(self) -> bool:
        """Сервер ведёт очередь: ставить версию можно только с его разрешения."""
        return self._watcher.reachable is True

    def is_granted(self, version: str) -> bool:
        with self._lock:
            return str(version) in self._granted_at

    def has_live_grant(self, now: float | None = None) -> bool:
        """Есть ли разрешение сервера, которое ещё действует (час с момента выдачи)."""
        moment = time.time() if now is None else now
        with self._lock:
            return any(0 <= moment - granted < GRANT_LIFETIME_S for granted in self._granted_at.values())

    def granted_at(self, version: str) -> float:
        with self._lock:
            return float(self._granted_at.get(str(version), 0.0))

    @staticmethod
    def busy_reason() -> str:
        """Почему обновляться сейчас не время; пусто — можно."""
        return "на экране игра или видео на весь экран" if fullscreen_app_active() else ""

    def set_report(self, report: dict | None) -> None:
        """Что сказать серверу об исходе прошлого обновления (один раз)."""
        with self._lock:
            self._report = {str(key): str(value) for key, value in (report or {}).items()}

    def note_miss(self, version: str, why: str) -> None:
        """Версия, которую программа взялась ставить сама, не скачалась.

        Серверу уходит ``miss`` с причиной: раздачу это не держит, но видно,
        почему у людей срывается скачивание. Неотправленный исход прошлого
        обновления («дошла», «не вышло») не затирается.
        """
        code = str(why or "").strip().lower()
        if not (version and WHY_RE.fullmatch(code)):
            return
        with self._lock:
            if not self._report:
                self._report = {"miss": str(version), "why": code}

    # --- для слушателя (его поток) ---

    def is_urgent(self, version: str) -> bool:
        with self._lock:
            return str(version) in self._urgent

    def _on_revoked(self, version: str) -> None:
        with self._lock:
            self._urgent.add(str(version))
        self._on_release(version)

    def _on_release(self, version: str) -> None:
        with self._lock:
            self._granted_at.setdefault(str(version), time.time())
        self.granted.emit(str(version))

    def _activity(self) -> dict:
        with self._lock:
            presence = dict(self._presence)
        if not presence:
            return {}
        if fullscreen_app_active():
            act = ACTIVITY_FULLSCREEN
        else:
            act = ACTIVITY_WINDOW if presence["window"] else ACTIVITY_TRAY
        # scr — сырой ответ Windows о занятости экрана, только общим счётом.
        return {"act": act, "run": "1" if presence["connected"] else "0", "scr": screen_state()}

    def _pending_report(self) -> dict:
        with self._lock:
            return dict(self._report)

    def _report_delivered(self, _report: dict) -> None:
        with self._lock:
            self._report = {}