""" Kwork API Client. Main client class with authentication and all API endpoints. """ import logging from typing import Any import httpx from .errors import ( KworkApiError, KworkAuthError, KworkError, KworkNetworkError, KworkNotFoundError, KworkRateLimitError, KworkValidationError, ) from .models import ( Badge, CatalogResponse, City, Country, Dialog, Feature, Kwork, KworkDetails, NotificationsResponse, Project, ProjectsResponse, ReviewsResponse, TimeZone, ValidationResponse, ) logger = logging.getLogger(__name__) class KworkClient: """ Асинхронный клиент для Kwork.ru API. Предоставляет доступ ко всем основным эндпоинтам Kwork API: - Каталог кворков и поиск - Биржа проектов (фриланс заказы) - Пользовательские данные и отзывы - Уведомления и сообщения - Справочные данные (города, страны, категории) Аутентификация: Клиент использует двухэтапную аутентификацию Kwork: 1. POST /signIn - получение session cookies 2. POST /getWebAuthToken - получение web_auth_token Примеры использования: # Вход по логину/паролю async with await KworkClient.login("username", "password") as client: catalog = await client.catalog.get_list(page=1) # Восстановление сессии по токену client = KworkClient(token="saved_web_auth_token") user_info = await client.user.get_info() # Работа с проектами projects = await client.projects.get_list(page=1) my_orders = await client.projects.get_payer_orders() Attributes: catalog: Доступ к каталогу кворков projects: Доступ к бирже проектов user: Пользовательские эндпоинты reference: Справочные данные notifications: Уведомления и сообщения other: Прочие эндпоинты Note: Клиент поддерживает context manager для автоматического закрытия соединения. Рекомендуется использовать `async with` для корректного освобождения ресурсов. """ BASE_URL = "https://api.kwork.ru" # HAR shows all API endpoints use api.kwork.ru def __init__( self, token: str | None = None, cookies: dict[str, str] | None = None, timeout: float = 30.0, base_url: str | None = None, ): """ Инициализация клиента. Создаёт неаутентифицированный клиент или восстанавливает сессию по ранее полученному токену. Args: token: Web auth token, полученный через `getWebAuthToken` или `login()`. Если указан, автоматически добавляется в cookies. cookies: Session cookies из предыдущей аутентификации. Обычно не требуется - устанавливаются автоматически из token. timeout: Таймаут HTTP запросов в секундах. По умолчанию 30 секунд. base_url: Кастомный базовый URL. Используется только для тестирования. Example: # Новый клиент без аутентификации client = KworkClient() # Восстановление сессии client = KworkClient(token="eyJ0eXAiOiJKV1QiLCJhbGc...") # Клиент с кастомным таймаутом client = KworkClient(timeout=60.0) Note: Для полноценной работы API требуется аутентификация. Используйте `login()` или передайте сохранённый token. """ self.base_url = base_url or self.BASE_URL self.timeout = timeout self._token = token self._cookies = cookies or {} # Initialize HTTP client self._client: httpx.AsyncClient | None = None @property def token(self) -> str | None: """ Web auth token для аутентификации. Returns: Токен или None если клиент не аутентифицирован. Example: # Сохранение токена для последующего использования client = await KworkClient.login("user", "pass") token = client.token # Позже: восстановление сессии client = KworkClient(token=token) """ return self._token @property def cookies(self) -> dict[str, str]: """ Session cookies. Returns: Словарь cookies включая web_auth_token. Example: # Сохранение полной сессии client = await KworkClient.login("user", "pass") creds = client.credentials # Восстановление client = KworkClient(**creds) """ return self._cookies.copy() @property def credentials(self) -> dict[str, str]: """ Учётные данные для восстановления сессии. Returns: Словарь со всеми cookies (включая slrememberme и userId) для передачи в KworkClient(cookies=...). Example: # Сохранение client = await KworkClient.login("user", "pass") import json with open("session.json", "w") as f: json.dump(client.credentials, f) # Восстановление with open("session.json") as f: creds = json.load(f) client = KworkClient(cookies=creds) """ return self._cookies.copy() if self._cookies else {} @classmethod async def login( cls, username: str, password: str, timeout: float = 30.0, ) -> "KworkClient": """ Аутентификация по логину и паролю. Выполняет двухэтапный процесс аутентификации Kwork: 1. POST /signIn - проверка учётных данных, получение session cookies 2. POST /getWebAuthToken - обмен cookies на web_auth_token Полученный токен и cookies сохраняются в клиенте для последующих запросов. Args: username: Логин или email аккаунта Kwork. password: Пароль аккаунта Kwork. timeout: Таймаут запросов в секундах. Применяется к каждому этапу. Returns: Полностью аутентифицированный экземпляр KworkClient, готовый к работе с API. Raises: KworkAuthError: Если логин/пароль неверны или токен не получен. KworkNetworkError: Если произошла ошибка сети. Example: # Базовое использование client = await KworkClient.login("myuser", "mypassword") # С кастомным таймаутом client = await KworkClient.login("user", "pass", timeout=60.0) # Сохранение токена для повторного использования token = client._token # Позже: client = KworkClient(token=token) Security: Пароль не сохраняется в клиенте. Только token и cookies. Рекомендуется сохранять token для повторного использования вместо хранения пароля. Note: Токен имеет ограниченное время жизни. При получении 401 ошибки необходимо выполнить повторный login(). """ client = cls(timeout=timeout) try: async with client._get_httpx_client() as http_client: # Step 1: Login to get session cookies and token # HAR analysis (mitmproxy + har-analyzer skill): # POST https://api.kwork.ru/signIn # Required headers: Authorization (Basic mobile_api:qFvfRl7w), User-Agent (Android), OS-Version # Content-Type: application/x-www-form-urlencoded login_data = { "login": username, "password": password, "uad": "", "device": "", } logger.info(f"Login request: POST https://api.kwork.ru/signIn (user: {username})") logger.debug(f"Login payload: {login_data}") response = await http_client.post( "https://api.kwork.ru/signIn", data=login_data, headers={ "Authorization": "Basic bW9iaWxlX2FwaTpxRnZmUmw3dw==", "User-Agent": "Kwork android client, version: 3.8.1", "OS-Version": "30", "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", }, ) logger.debug(f"Login response status: {response.status_code}") logger.debug(f"Login response headers: {dict(response.headers)}") if response.status_code != 200: logger.error( f"Login failed with status {response.status_code}: {response.text[:200]}" ) raise KworkAuthError(f"Login failed: {response.status_code}") response_data = response.json() cookies = dict(response.cookies) logger.info( f"Login successful: user_id={cookies.get('userId')}, csrf_token={response_data.get('csrftoken', 'N/A')[:20] if response_data.get('csrftoken') else 'N/A'}" ) logger.debug(f"Login response data: {response_data}") logger.debug(f"Login cookies: {list(cookies.keys())}") # Extract userId from cookies user_id = cookies.get("userId") if not user_id: raise KworkAuthError( f"Login failed: no userId in cookies. Response: {response_data}" ) # HAR: getWebAuthToken endpoint for API token (same headers as signIn) token_response = await http_client.post( "https://api.kwork.ru/getWebAuthToken", json={}, headers={ "Authorization": "Basic bW9iaWxlX2FwaTpxRnZmUmw3dw==", "User-Agent": "Kwork android client, version: 3.8.1", "OS-Version": "30", "Accept": "application/json", }, ) if token_response.status_code != 200: raise KworkAuthError(f"Token request failed: {token_response.status_code}") token_data = token_response.json() # HAR shows: {"success":true,"response":{"token":"...", "expires_at":..., "url":...}} web_token = token_data.get("response", {}).get("token") if not web_token: raise KworkAuthError(f"No token in response: {token_data}") logger.info(f"Got web_auth_token: {web_token[:20]}...") # Create new client with token return cls(token=web_token, cookies=cookies, timeout=timeout) except httpx.RequestError as e: raise KworkNetworkError(f"Login request failed: {e}") from e def _get_httpx_client(self) -> httpx.AsyncClient: """Get or create HTTP client with proper headers.""" if self._client is None or self._client.is_closed: headers = { "Accept": "application/json", "Content-Type": "application/json", "Referer": "https://kwork.ru/", "Origin": "https://kwork.ru", # HAR: All API requests require Authorization header "Authorization": "Basic bW9iaWxlX2FwaTpxRnZmUmw3dw==", "User-Agent": "Kwork android client, version: 3.8.1", "OS-Version": "30", } if self._token: # HAR: API requires both Authorization header AND slrememberme cookie self._cookies["slrememberme"] = self._token # Convert cookies to Cookie header string for cross-domain compatibility if self._cookies: cookie_header = "; ".join(f"{k}={v}" for k, v in self._cookies.items()) headers["Cookie"] = cookie_header logger.debug(f"Setting Cookie header: {cookie_header[:100]}...") self._client = httpx.AsyncClient( base_url=self.base_url, headers=headers, timeout=self.timeout, http2=True, ) return self._client async def close(self) -> None: """Close HTTP client.""" if self._client and not self._client.is_closed: await self._client.aclose() async def __aenter__(self) -> "KworkClient": return self async def __aexit__(self, *args: Any) -> None: await self.close() def _handle_response(self, response: httpx.Response) -> dict[str, Any]: """ Handle HTTP response and raise appropriate errors. Args: response: HTTP response Returns: Response JSON data Raises: KworkApiError: For HTTP errors KworkAuthError: For auth errors KworkNetworkError: For network errors """ # Check for common error statuses if response.status_code == 401: raise KworkAuthError("Unauthorized: invalid or expired token") if response.status_code == 403: raise KworkAuthError("Forbidden: access denied") if response.status_code == 404: raise KworkNotFoundError(f"Resource not found: {response.url}") if response.status_code == 429: raise KworkRateLimitError("Too many requests") if response.status_code >= 400: try: error_data = response.json() message = error_data.get("message", str(error_data)) except Exception: message = response.text if response.status_code == 400: raise KworkValidationError(message, response=response) raise KworkApiError(message, response.status_code, response) # Parse successful response try: return response.json() except Exception as e: raise KworkError(f"Failed to parse response: {e}") from e async def _request( self, method: str, endpoint: str, **kwargs: Any, ) -> dict[str, Any]: """ Make HTTP request. Args: method: HTTP method endpoint: API endpoint **kwargs: Additional arguments for httpx Returns: Response JSON data """ http_client = self._get_httpx_client() # Log request full_url = f"{self.base_url}{endpoint}" if not endpoint.startswith("http") else endpoint logger.debug(f"API Request: {method} {full_url}") logger.debug(f"Request kwargs: {kwargs}") try: response = await http_client.request(method, endpoint, **kwargs) # Log response logger.debug(f"API Response: {response.status_code} {response.reason_phrase}") logger.debug(f"Response headers: {dict(response.headers)}") # Log response body (truncated for large responses) try: response_text = response.text if len(response_text) > 500: logger.debug(f"Response body (truncated): {response_text[:500]}...") else: logger.debug(f"Response body: {response_text}") except Exception: logger.debug("Response body: ") return self._handle_response(response) except httpx.RequestError as e: logger.error(f"Network error for {method} {full_url}: {e}") raise KworkNetworkError(f"Request failed: {e}") from e # ========== Catalog Endpoints ========== class CatalogAPI: """ API каталога кворков. Предоставляет доступ к каталогу услуг Kwork: - Поиск и фильтрация кворков - Получение детальной информации - Категории и сортировка Example: # Получить первую страницу каталога catalog = await client.catalog.get_list(page=1) # Фильтрация по категории catalog = await client.catalog.get_list(category_id=5) # Детали конкретного кворка details = await client.catalog.get_details(kwork_id=12345) """ def __init__(self, client: "KworkClient"): self.client = client async def get_list( self, page: int = 1, category_id: int | None = None, sort: str = "recommend", ) -> CatalogResponse: """ Получить список кворков из каталога. Основной эндпоинт для поиска и просмотра кворков. Возвращает пагинированный список с возможностью фильтрации. Args: page: Номер страницы для пагинации (начиная с 1). category_id: ID категории для фильтрации. Если None - все категории. sort: Опция сортировки. Варианты: - "recommend" - по рекомендации (по умолчанию) - "price_asc" - по возрастанию цены - "price_desc" - по убыванию цены - "rating" - по рейтингу - "reviews" - по количеству отзывов - "newest" - по дате создания Returns: CatalogResponse содержащий: - kworks: список кворков на странице - pagination: информация о пагинации - filters: доступные фильтры - sort_options: доступные опции сортировки Example: # Первая страница, сортировка по цене response = await client.catalog.get_list( page=1, sort="price_asc" ) for kwork in response.kworks: print(f"{kwork.title}: {kwork.price} RUB") # Пагинация if response.pagination and response.pagination.has_next: next_page = await client.catalog.get_list(page=2) """ data = await self.client._request( "POST", "/catalogMainv2", # TODO: 404 - need to find correct endpoint (HAR shows GET /categories/{slug}) json={ "page": page, "category_id": category_id, "sort": sort, }, ) return CatalogResponse.model_validate(data) async def get_details(self, kwork_id: int) -> KworkDetails: """ Получить полную информацию о кворке. Возвращает расширенную информацию о кворке включая: - Полное описание и требования - Сроки выполнения и количество правок - Дополнительные опции (features) - FAQ от продавца Args: kwork_id: Уникальный идентификатор кворка. Returns: KworkDetails с полной информацией о кворке. Raises: KworkNotFoundError: Если кворк с таким ID не найден. Example: details = await client.catalog.get_details(12345) print(f"Название: {details.title}") print(f"Цена: {details.price} {details.currency}") print(f"Срок: {details.delivery_time} дней") print(f"Правки: {details.revisions}") """ data = await self.client._request( "POST", "/getKworkDetails", # TODO: 404 - HAR shows GET /projects/{id}/view json={"kwork_id": kwork_id}, ) return KworkDetails.model_validate(data) async def get_details_extra(self, kwork_id: int) -> dict[str, Any]: """ Получить дополнительные детали кворка. Возвращает расширенную информацию, которая не включена в основной ответ get_details(). Может содержать: - Дополнительные изображения - Видео обзоры - Детали пакетов услуг - Статистику продаж Args: kwork_id: Уникальный идентификатор кворка. Returns: Словарь с дополнительными данными. Структура зависит от конкретного кворка и не гарантируется стабильной. Note: Этот эндпоинт возвращает "сырые" данные без валидации. Структура ответа может измениться без предупреждения. """ return await self.client._request( "POST", "/getKworkDetailsExtra", json={"kwork_id": kwork_id}, ) # ========== Projects Endpoints ========== class ProjectsAPI: """ API биржи проектов (фриланс заказы). Предоставляет доступ к заказам на фриланс: - Просмотр открытых проектов - Заказы где вы заказчик (payer) - Заказы где вы исполнитель (worker) Example: # Новые проекты projects = await client.projects.get_list(page=1) # Мои заказы как заказчика my_orders = await client.projects.get_payer_orders() # Мои заказы как исполнителя my_work = await client.projects.get_worker_orders() """ def __init__(self, client: "KworkClient"): self.client = client async def get_list( self, page: int = 1, category_id: int | None = None, ) -> ProjectsResponse: """ Получить список проектов с биржи. Основной эндпоинт для просмотра доступных заказов. Возвращает пагинированный список проектов. Args: page: Номер страницы (начиная с 1). category_id: ID категории для фильтрации. Если None - все категории. Returns: ProjectsResponse содержащий: - projects: список проектов на странице - pagination: информация о пагинации Example: # Все новые проекты response = await client.projects.get_list(page=1) for project in response.projects: print(f"{project.title}: {project.budget} {project.budget_type}") # Только категория "Разработка" dev_projects = await client.projects.get_list( page=1, category_id=5 ) """ data = await self.client._request( "POST", "/projects", json={ "page": page, "category_id": category_id, }, ) return ProjectsResponse.model_validate(data) async def get_payer_orders(self) -> list[Project]: """ Получить заказы где вы являетесь заказчиком. Возвращает все проекты, созданные текущим пользователем, независимо от их статуса (открыт, в работе, завершён). Returns: Список проектов где текущий пользователь - заказчик. Example: orders = await client.projects.get_payer_orders() for order in orders: print(f"Заказ #{order.id}: {order.status}") """ data = await self.client._request("POST", "/payerOrders") return [Project.model_validate(p) for p in data.get("orders", [])] async def get_worker_orders(self) -> list[Project]: """ Получить заказы где вы являетесь исполнителем. Возвращает все проекты, где текущий пользователь назначен исполнителем. Returns: Список проектов где текущий пользователь - исполнитель. Example: work = await client.projects.get_worker_orders() active = [p for p in work if p.status == "in_progress"] print(f"Активных заказов: {len(active)}") """ data = await self.client._request("POST", "/workerOrders") return [Project.model_validate(p) for p in data.get("orders", [])] # ========== User Endpoints ========== class UserAPI: """ Пользовательское API. Предоставляет доступ к данным текущего пользователя: - Профиль и информация об аккаунте - Отзывы (полученные и оставленные) - Избранные кворки Example: # Информация о пользователе info = await client.user.get_info() # Мои отзывы reviews = await client.user.get_reviews() # Избранные кворки favorites = await client.user.get_favorite_kworks() """ def __init__(self, client: "KworkClient"): self.client = client async def get_info(self) -> dict[str, Any]: """ Получить информацию о текущем пользователе. Возвращает основные данные аккаунта: - ID, username, email - Баланс, рейтинг - Статус верификации - Настройки профиля Returns: Словарь с информацией о пользователе. Структура зависит от ответа API. Example: info = await client.user.get_info() print(f"User: {info.get('username')}") print(f"Balance: {info.get('balance')} RUB") """ # HAR: POST https://api.kwork.ru/user return await self.client._request("POST", "/user") async def get_reviews( self, user_id: int | None = None, page: int = 1, ) -> ReviewsResponse: """ Получить отзывы пользователя. Если user_id не указан - возвращает отзывы текущего пользователя. Если указан - отзывы другого пользователя по ID. Args: user_id: ID пользователя. Если None - текущий пользователь. page: Номер страницы для пагинации (начиная с 1). Returns: ReviewsResponse содержащий: - reviews: список отзывов на странице - pagination: информация о пагинации - average_rating: средний рейтинг Example: # Мои отзывы my_reviews = await client.user.get_reviews() # Отзывы другого пользователя user_reviews = await client.user.get_reviews(user_id=12345) # С пагинацией page2 = await client.user.get_reviews(page=2) """ # HAR: POST https://api.kwork.ru/userReviews data = await self.client._request( "POST", "/userReviews", json={"user_id": user_id, "page": page} if user_id else {"page": page}, ) return ReviewsResponse.model_validate(data) async def get_favorite_kworks(self) -> list[Kwork]: """ Получить список избранных кворков. Возвращает все кворки, добавленные пользователем в избранное. Returns: Список избранных кворков. Example: favorites = await client.user.get_favorite_kworks() for kwork in favorites: print(f"{kwork.title}: {kwork.price} RUB") """ data = await self.client._request("POST", "/favoriteKworks") return [Kwork.model_validate(k) for k in data.get("kworks", [])] # ========== Reference Data Endpoints ========== class ReferenceAPI: """ Справочное API. Предоставляет доступ к справочным данным Kwork: - Города, страны, часовые пояса - Доступные функции и дополнения - Значки пользователей Эти данные редко меняются и могут быть закэшированы. Example: # Все страны countries = await client.reference.get_countries() # Все города cities = await client.reference.get_cities() # Доступные фичи features = await client.reference.get_features() """ def __init__(self, client: "KworkClient"): self.client = client async def get_cities(self) -> list[City]: """ Получить список всех городов. Returns: Список всех городов из справочника Kwork. Example: cities = await client.reference.get_cities() moscow = next(c for c in cities if c.name == "Москва") """ # TODO: 404 - endpoint not found in HAR, may need to parse HTML or find JS data data = await self.client._request("POST", "/cities") return [City.model_validate(c) for c in data.get("cities", [])] async def get_countries(self) -> list[Country]: """ Получить список всех стран. Returns: Список всех стран с кодами и городами. Example: countries = await client.reference.get_countries() russia = next(c for c in countries if c.code == "RU") """ # TODO: 404 - endpoint not found in HAR, may need to parse HTML or find JS data data = await self.client._request("POST", "/countries") return [Country.model_validate(c) for c in data.get("countries", [])] async def get_timezones(self) -> list[TimeZone]: """ Получить список всех часовых поясов. Returns: Список часовых поясов с названиями и смещениями. Example: timezones = await client.reference.get_timezones() msks = next(tz for tz in timezones if "Moscow" in tz.name) """ data = await self.client._request("POST", "/timezones") return [TimeZone.model_validate(t) for t in data.get("timezones", [])] async def get_features(self) -> list[Feature]: """ Получить доступные дополнительные функции (features). Features - это платные дополнения к кворкам: - Увеличенные сроки - Дополнительные правки - Приоритетная поддержка - и т.д. Returns: Список доступных features с названиями и ценами. Example: features = await client.reference.get_features() for f in features: print(f"{f.name}: {f.price} RUB") """ data = await self.client._request("POST", "/getAvailableFeatures") return [Feature.model_validate(f) for f in data.get("features", [])] async def get_public_features(self) -> list[Feature]: """ Получить публичные функции. Аналогично get_features(), но возвращает только публично доступные опции. Returns: Список публичных features. """ data = await self.client._request("POST", "/getPublicFeatures") return [Feature.model_validate(f) for f in data.get("features", [])] async def get_badges_info(self) -> list[Badge]: """ Получить информацию о значках пользователей. Значки (badges) отображают достижения и статусы: - "Профессионал" - "Быстрый ответ" - "Надёжный продавец" - и т.д. Returns: Список значков с описаниями и иконками. Example: badges = await client.reference.get_badges_info() for badge in badges: print(f"{badge.name}: {badge.description}") """ data = await self.client._request("POST", "/getBadgesInfo") return [Badge.model_validate(b) for b in data.get("badges", [])] # ========== Notifications & Messages ========== class NotificationsAPI: """ API уведомлений и сообщений. Предоставляет доступ к системе уведомлений Kwork: - Список уведомлений - Получение новых уведомлений - Диалоги с пользователями - Заблокированные диалоги Example: # Все уведомления notifications = await client.notifications.get_list() # Новые уведомления new = await client.notifications.fetch() # Диалоги dialogs = await client.notifications.get_dialogs() """ def __init__(self, client: "KworkClient"): self.client = client async def get_list(self) -> NotificationsResponse: """ Получить список всех уведомлений. Возвращает все уведомления пользователя с информацией о прочтении. Returns: NotificationsResponse содержащий: - notifications: список уведомлений - unread_count: количество непрочитанных Example: notifs = await client.notifications.get_list() print(f"Непрочитанных: {notifs.unread_count}") for n in notifs.notifications: if not n.is_read: print(f"Новое: {n.title}") """ data = await self.client._request("POST", "/notifications") return NotificationsResponse.model_validate(data) async def fetch(self) -> NotificationsResponse: """ Получить новые уведомления. Отличается от get_list() тем, что возвращает только уведомления, появившиеся с момента последнего запроса. Также может обновлять счётчик непрочитанных. Returns: NotificationsResponse с новыми уведомлениями. Example: new_notifs = await client.notifications.fetch() if new_notifs.unread_count > 0: print(f"У вас {new_notifs.unread_count} новых уведомлений!") """ data = await self.client._request("POST", "/notificationsFetch") return NotificationsResponse.model_validate(data) async def get_dialogs(self) -> list[Dialog]: """ Получить список диалогов (чатов). Возвращает все активные диалоги пользователя с другими пользователями Kwork. Returns: Список диалогов с последней перепиской. Example: dialogs = await client.notifications.get_dialogs() for d in dialogs: print(f"С {d.participant.username}: {d.last_message}") """ data = await self.client._request("POST", "/dialogs") return [Dialog.model_validate(d) for d in data.get("dialogs", [])] async def get_blocked_dialogs(self) -> list[Dialog]: """ Получить список заблокированных диалогов. Возвращает диалоги с пользователями, которые были заблокированы текущим пользователем. Returns: Список заблокированных диалогов. Example: blocked = await client.notifications.get_blocked_dialogs() print(f"Заблокировано: {len(blocked)} пользователей") """ data = await self.client._request("POST", "/blockedDialogList") return [Dialog.model_validate(d) for d in data.get("dialogs", [])] # ========== Other Endpoints ========== class OtherAPI: """ Прочее API. Вспомогательные эндпоинты которые не вошли в другие категории: - Пользовательские настройки и предпочтения (wants) - Статусы кворков и заказов - Персональные предложения (offers) - Настройки профиля - Статус онлайн/оффлайн Example: # Пользовательские предпочтения wants = await client.other.get_wants() # Статус кворков status = await client.other.get_kworks_status() # Установить статус оффлайн await client.other.go_offline() """ def __init__(self, client: "KworkClient"): self.client = client async def get_wants(self) -> dict[str, Any]: """ Получить пользовательские предпочтения (wants). Wants - это настройки интересов пользователя: - Предпочитаемые категории - Ключевые слова для мониторинга - Фильтры для поиска Returns: Словарь с настройками предпочтений. Example: wants = await client.other.get_wants() print(wants) """ return await self.client._request("POST", "/myWants") async def get_wants_status(self) -> dict[str, Any]: """ Получить статус предпочтений. Returns: Статус wants с метаданными. """ return await self.client._request("POST", "/wantsStatusList") async def get_kworks_status(self) -> dict[str, Any]: """ Получить статусы кворков пользователя. Возвращает информацию о статусах всех кворков текущего пользователя (активен, на модерации, и т.д.). Returns: Словарь со статусами кворков. Example: status = await client.other.get_kworks_status() print(status) """ return await self.client._request("POST", "/kworksStatusList") async def get_offers(self) -> dict[str, Any]: """ Получить персональные предложения. Returns: Список персональных предложений от Kwork. """ return await self.client._request("POST", "/offers") async def get_exchange_info(self) -> dict[str, Any]: """ Получить информацию об обмене валюты. Returns: Информация о курсах валют и обмене. """ return await self.client._request("POST", "/exchangeInfo") async def get_channel(self) -> dict[str, Any]: """ Получить информацию о канале пользователя. Returns: Данные канала (если есть). """ return await self.client._request("POST", "/getChannel") async def get_in_app_notification(self) -> dict[str, Any]: """ Получить внутриприложенное уведомление. Returns: Данные in-app уведомления. """ return await self.client._request("POST", "/getInAppNotification") async def get_security_user_data(self) -> dict[str, Any]: """ Получить данные безопасности пользователя. Returns: Информация о безопасности аккаунта. """ return await self.client._request("POST", "/getSecurityUserData") async def is_dialog_allow(self, user_id: int) -> bool: """ Проверить возможность начала диалога с пользователем. Args: user_id: ID пользователя для проверки. Returns: True если диалог разрешён, False иначе. Example: allowed = await client.other.is_dialog_allow(12345) if allowed: print("Можно написать сообщение") """ data = await self.client._request( "POST", "/isDialogAllow", json={"user_id": user_id}, ) return data.get("allowed", False) async def get_viewed_kworks(self) -> list[Kwork]: """ Получить просмотренные кворки. Возвращает список кворков, которые пользователь просматривал ранее. Returns: Список просмотренных кворков. Example: viewed = await client.other.get_viewed_kworks() print(f"Просмотрено: {len(viewed)} кворков") """ data = await self.client._request("POST", "/viewedCatalogKworks") return [Kwork.model_validate(k) for k in data.get("kworks", [])] async def get_favorite_categories(self) -> list[int]: """ Получить ID избранных категорий. Returns: Список ID категорий, добавленных в избранное. Example: cats = await client.other.get_favorite_categories() print(f"Избранные категории: {cats}") """ data = await self.client._request("POST", "/favoriteCategories") return data.get("categories", []) async def update_settings(self, settings: dict[str, Any]) -> dict[str, Any]: """ Обновить настройки пользователя. Args: settings: Словарь с настройками для обновления. Структура зависит от конкретных настроек. Returns: Ответ API с подтверждением обновления. Example: await client.other.update_settings({ "email_notifications": True, "language": "ru" }) """ return await self.client._request("POST", "/updateSettings", json=settings) async def go_offline(self) -> dict[str, Any]: """ Установить статус пользователя "оффлайн". Скрывает онлайн-статус от других пользователей. Returns: Подтверждение изменения статуса. Example: await client.other.go_offline() """ return await self.client._request("POST", "/offline") async def get_actor(self) -> dict[str, Any]: """ Получить информацию об актёре (текущем пользователе). Returns: Данные актёра/пользователя. """ return await self.client._request("POST", "/actor") async def validate_text(self, text: str, context: str | None = None) -> ValidationResponse: """ Проверить текст на соответствие требованиям Kwork. Эндпоинт валидации текста используется для проверки: - Описаний кворков - Текстов проектов - Сообщений и отзывов Находит потенциальные проблемы: - Запрещённые слова и контакты - Нарушения правил площадки - Проблемы с форматированием Args: text: Текст для проверки. context: Контекст использования (опционально). Например: "kwork_description", "project_text", "message". Returns: ValidationResponse с результатами валидации. Example: result = await client.other.validate_text( "Отличный сервис, звоните +7-999-000-00-00" ) if not result.is_valid: print("Текст не прошёл валидацию:") for issue in result.issues: print(f" - {issue.message}") """ payload = {"text": text} if context: payload["context"] = context data = await self.client._request( "POST", "/api/validation/checktext", json=payload, ) return ValidationResponse.model_validate(data) # ========== API Property Accessors ========== @property def catalog(self) -> CatalogAPI: """API каталога кворков.""" return self.CatalogAPI(self) @property def projects(self) -> ProjectsAPI: """API биржи проектов.""" return self.ProjectsAPI(self) @property def user(self) -> UserAPI: """Пользовательское API.""" return self.UserAPI(self) @property def reference(self) -> ReferenceAPI: """Справочное API.""" return self.ReferenceAPI(self) @property def notifications(self) -> NotificationsAPI: """API уведомлений.""" return self.NotificationsAPI(self) @property def other(self) -> OtherAPI: """Прочее API.""" return self.OtherAPI(self)