BitrixDealsBot/apps/bot/bitrix.py

import asyncio
from datetime import UTC, datetime, timedelta
from typing import Any

import httpx

from .crypto import TokenCipher
from .domain import Binding, OAuthCredentials
from .oauth import BotOAuthRepository

AUTH_ERRORS = {"expired_token", "invalid_token", "no_auth_found"}


class BitrixClient:
    """REST-клиент Битрикса с OAuth-контекстом привязанного пользователя."""

    def __init__(
            self,
            credentials: BotOAuthRepository,
            cipher: TokenCipher,
            client_id: str,
            client_secret: str,
            oauth_token_url: str,
            client: httpx.AsyncClient | None = None
    ) -> None:
        self.credentials = credentials
        self.cipher = cipher
        self.client_id = client_id
        self.client_secret = client_secret
        self.oauth_token_url = oauth_token_url
        # Передача клиента для упрощения тестирования.
        self._client = client or httpx.AsyncClient(timeout=15)
        self._owns_client = client is None

    async def call(
            self,
            binding: Binding,
            method: str,
            params: dict[str, Any] | None = None
    ) -> dict[str, Any]:
        credentials = await self.credentials.get(binding)
        if not credentials:
            raise RuntimeError("OAuth-данные пользователя не найдены")

        # Отправляем запрос и в случае истечения токена запрашиваем обновление.
        data = await self._request(credentials, method, params)
        if str(data.get("error") or "").lower() in AUTH_ERRORS:
            credentials = await self._refresh(credentials, binding)
            data = await self._request(credentials, method, params)

        if "error" in data:
            description = data.get("error_description", data["error"])
            raise RuntimeError(f"Bitrix API error: {description}")
        return data

    async def _request(
            self,
            credentials: OAuthCredentials,
            method: str,
            params: dict[str, Any] | None
    ) -> dict[str, Any]:
        payload = dict(params or {})
        payload["auth"] = self.cipher.decrypt(credentials.access_token)
        response = await self._client.post(
            f"https://{credentials.domain}/rest/{method}.json",
            json=payload
        )
        try:
            data = response.json()
        except ValueError:
            response.raise_for_status()
            raise RuntimeError("Bitrix вернул некорректный ответ") from None

        # Битрикс присылает полезное описание ошибки и при HTTP 4xx.
        if response.is_error and "error" not in data:
            response.raise_for_status()
        return data

    async def _refresh(
            self,
            credentials: OAuthCredentials,
            binding: Binding
    ) -> OAuthCredentials:
        # Если другой процесс уже обновляет токен, ждем его завершения.
        if not await self.credentials.claim_refresh(credentials):
            return await self._wait_for_refresh(credentials, binding)

        try:
            # Битрикс возвращает новую пару, поэтому обновляем оба токена.
            try:
                response = await self._client.get(
                    self.oauth_token_url,
                    params={
                        "grant_type": "refresh_token",
                        "client_id": self.client_id,
                        "client_secret": self.client_secret,
                        "refresh_token": self.cipher.decrypt(
                            credentials.refresh_token)
                    }
                )
                response.raise_for_status()
            except httpx.HTTPError:
                # Не включаем URL с OAuth-секретами в traceback.
                raise RuntimeError("Не удалось обновить OAuth-токен") from None

            data = response.json()
            if "error" in data:
                raise RuntimeError(
                    "Bitrix OAuth error: "
                    + str(data.get("error_description") or data["error"])
                )

            # Проверяем, что обновленный токен принадлежит тому же порталу
            # и пользователю.
            if data.get("member_id") not in {None, credentials.member_id}:
                raise RuntimeError("Bitrix вернул токен другого портала")
            if int(data.get("user_id", credentials.bitrix_user_id)) != (
                    credentials.bitrix_user_id
            ):
                raise RuntimeError("Bitrix вернул токен другого пользователя")

            # Обновляем токены в базе и возвращаем новые данные.
            expires_at = datetime.now(UTC) + timedelta(
                seconds=int(data.get("expires_in", 3600))
            )
            saved = await self.credentials.finish_refresh(
                credentials,
                self.cipher.encrypt(str(data["access_token"])),
                self.cipher.encrypt(str(data["refresh_token"])),
                expires_at
            )

            # Если другой процесс успел обновить токен, ждем его завершения.
            if not saved:
                return await self._wait_for_refresh(credentials, binding)

            updated = await self.credentials.get(binding)
            if not updated:
                raise RuntimeError("Обновленные OAuth-данные не найдены")
            return updated

        except Exception:
            await self.credentials.release_refresh(credentials)
            raise

    async def _wait_for_refresh(
            self,
            previous: OAuthCredentials,
            binding: Binding
    ) -> OAuthCredentials:
        for _ in range(80):
            await asyncio.sleep(0.2)
            current = await self.credentials.get(binding)
            if current and current.version > previous.version:
                return current

        raise RuntimeError("Не удалось дождаться обновления OAuth-токена")

    async def close(self) -> None:
        if self._owns_client:
            await self._client.aclose()