93 lines
3.4 KiB
Python
93 lines
3.4 KiB
Python
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}")
|