bulid 63

hockey_data/client.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+from .config import HockeySettingsStore
+
+
+class Stat2TVNotFoundError(RuntimeError):
+    def __init__(self, url: str, status_code: int = 404) -> None:
+        self.url = url
+        self.status_code = status_code
+        super().__init__(f"Stat2TV resource not found: HTTP {status_code}")
+
+
+@dataclass(slots=True)
+class Stat2TVResponse:
+    content: bytes
+    status_code: int
+    content_type: str
+    auth_mode: str
+    url: str
+
+
+class Stat2TVClient:
+    def __init__(self, settings: HockeySettingsStore) -> None:
+        self.settings_store = settings
+        self._working_auth_mode: str | None = None
+
+    async def fetch_xml(self, endpoint: str) -> Stat2TVResponse:
+        settings = self.settings_store.public()
+        username, password = self.settings_store.credentials()
+        if not username or not password:
+            raise RuntimeError(
+                "Логин и пароль Stat2TV не настроены. "
+                "Откройте настройки хоккейного модуля."
+            )
+
+        url = settings["base_url"].rstrip("/") + "/" + endpoint.lstrip("/")
+        requested_mode = settings["auth_mode"]
+        if requested_mode in {"basic", "digest"}:
+            modes = [requested_mode]
+        elif self._working_auth_mode in {"basic", "digest"}:
+            other = "digest" if self._working_auth_mode == "basic" else "basic"
+            modes = [self._working_auth_mode, other]
+        else:
+            modes = ["basic", "digest"]
+        last_error: Exception | None = None
+
+        for mode in modes:
+            auth: Any = (
+                httpx.BasicAuth(username, password)
+                if mode == "basic"
+                else httpx.DigestAuth(username, password)
+            )
+            try:
+                async with httpx.AsyncClient(
+                    timeout=float(settings["request_timeout_seconds"]),
+                    verify=bool(settings["verify_ssl"]),
+                    follow_redirects=True,
+                    headers={
+                        "Accept": "application/xml,text/xml,*/*",
+                        "User-Agent": "HockeyControlPanel/20.2",
+                    },
+                ) as client:
+                    response = await client.get(url, auth=auth)
+                if response.status_code in {401, 403} and requested_mode == "auto":
+                    last_error = RuntimeError(
+                        f"Авторизация {mode} отклонена: HTTP {response.status_code}"
+                    )
+                    continue
+                if response.status_code in {404, 410}:
+                    raise Stat2TVNotFoundError(str(response.url), response.status_code)
+                response.raise_for_status()
+                self._working_auth_mode = mode
+                return Stat2TVResponse(
+                    content=response.content,
+                    status_code=response.status_code,
+                    content_type=response.headers.get("content-type", ""),
+                    auth_mode=mode,
+                    url=str(response.url),
+                )
+            except Stat2TVNotFoundError:
+                raise
+            except Exception as error:
+                last_error = error
+                if requested_mode != "auto":
+                    break
+
+        raise RuntimeError(f"Stat2TV недоступен: {last_error}")