98 lines
4.1 KiB
Python
98 lines
4.1 KiB
Python
import os
|
|
import requests
|
|
import time
|
|
from typing import Dict, Any, Optional
|
|
from .exceptions import AvitoAPIError, AvitoAuthError, AvitoRateLimitError, AvitoBadRequestError, AvitoNotFoundError
|
|
|
|
class AvitoClient:
|
|
"""Base client for interacting with Avito API."""
|
|
|
|
BASE_URL = "https://api.avito.ru"
|
|
|
|
def __init__(self, client_id: Optional[str] = None, client_secret: Optional[str] = None):
|
|
# Fetch from env if not provided
|
|
self.client_id = client_id or os.getenv('AVITO_CLIENT_ID')
|
|
self.client_secret = client_secret or os.getenv('AVITO_CLIENT_SECRET')
|
|
|
|
if not self.client_id or not self.client_secret:
|
|
raise ValueError("AVITO_CLIENT_ID and AVITO_CLIENT_SECRET must be provided")
|
|
|
|
self.session = requests.Session()
|
|
self.access_token = None
|
|
|
|
# Получаем токен сразу при инициализации
|
|
self._authenticate()
|
|
|
|
def _authenticate(self):
|
|
"""Exchange Client ID and Secret for a temporary OAuth access token."""
|
|
url = f"{self.BASE_URL}/token/"
|
|
data = {
|
|
"client_id": self.client_id,
|
|
"client_secret": self.client_secret,
|
|
"grant_type": "client_credentials"
|
|
}
|
|
|
|
response = requests.post(url, data=data)
|
|
if response.status_code != 200:
|
|
raise AvitoAuthError(f"Failed to get Avito token: {response.text}")
|
|
|
|
token_data = response.json()
|
|
self.access_token = token_data.get("access_token")
|
|
|
|
self.session.headers.update({
|
|
"Authorization": f"Bearer {self.access_token}",
|
|
"Content-Type": "application/json"
|
|
})
|
|
|
|
def request(self, method: str, path: str, data: Optional[Dict[str, Any]] = None, params: Optional[Dict[str, Any]] = None, retries: int = 3) -> Dict[str, Any]:
|
|
"""Perform HTTP request to Avito API with automatic retries and token refresh."""
|
|
url = f"{self.BASE_URL}{path}"
|
|
|
|
for attempt in range(retries):
|
|
try:
|
|
response = self.session.request(method, url, json=data, params=params)
|
|
|
|
# Check for expired token (401) or Forbidden (403)
|
|
if response.status_code in (401, 403):
|
|
if attempt < retries - 1:
|
|
# Обновляем токен и пробуем снова
|
|
self._authenticate()
|
|
continue
|
|
raise AvitoAuthError(f"Authentication failed after retries: {response.text}")
|
|
|
|
# Check rate limiting (429)
|
|
if response.status_code == 429:
|
|
if attempt < retries - 1:
|
|
time.sleep(2 ** attempt) # Exponential backoff
|
|
continue
|
|
raise AvitoRateLimitError("Rate limit exceeded")
|
|
|
|
if response.status_code == 400:
|
|
raise AvitoBadRequestError(f"Bad Request: {response.text}")
|
|
|
|
if response.status_code == 404:
|
|
raise AvitoNotFoundError(f"Not Found: {response.text}")
|
|
|
|
response.raise_for_status()
|
|
|
|
if response.content:
|
|
return response.json()
|
|
return {}
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
if attempt < retries - 1:
|
|
time.sleep(2 ** attempt)
|
|
continue
|
|
raise AvitoAPIError(f"Request failed: {str(e)}")
|
|
|
|
raise AvitoAPIError("Request failed after max retries")
|
|
|
|
def post(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
return self.request("POST", path, data=data)
|
|
|
|
def get(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
return self.request("GET", path, params=params)
|
|
|
|
def put(self, path: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
|
return self.request("PUT", path, data=data)
|