PyPlayerokAPI 1.0.1

Неофициальная асинхронная Python-библиотека для взаимодействия с торговой площадкой Playerok через GraphQL API и систему потоковых событий.

Асинхронный Python SDK для работы с Playerok API
GraphQL • Streaming • Proxy • Production-ready transport

---

PyPlayerokAPI — неофициальная Python-библиотека для взаимодействия с торговой площадкой Playerok через GraphQL API и систему потоковых событий.

Что можно сделать с помощью библиотеки:

• 🤖 автоматизацию
• 📊 мониторинг сделок
• 💬 работы с чатами
• 💰 управления транзакциями
• 🔄 обработки событий в реальном времени
• 🧩 интеграции в собственные сервисы

---

📚 Содержание

• [[#Особенности]]
• [[#Требования]]
• [[#Установка]]
• [[#Аутентификация]]
• [[#Структура библиотеки]]
• [[#Примеры использования]]
  • [[#Предисловие]]
  • [[#Общие варианты использования]]
  • [[#Готовые варианты использования]]
• [[#Дополнительная информация]]

---

Особенности

• Выполнение запросов GraphQL
• Поддержка Persisted queries
• Модуль управления аккаунтом
• Стриминг & обработка событий в реальном времени
• Поддержка прокси
• Аутентификация на основе токена
• Кастомный транспортный уровень, совместимый с Cloudflare (через tls_requests и curl_cffi)

---

Требования

• Python 3.11+
• token аккаунта Playerok
• Прокси (HTTP/S) (опционально)

---

Установка

Через PyPI

pip install pyplayerokapi

С помощью pip

pip install git+https://github.com/kekch127/PyPlayerokAPI.git

Из репозитория

git clone https://github.com/kekch127/PyPlayerokAPI.git
cd PyPlayerokAPI
pip install -e .

---

Аутентификация

Для использования необходимо сначала получить токен аккаунта Playerok. Для этого авторизуйтесь на сайте, после с помощью любого расширения (я использую EditThisCookie V3) получите JWT токен из Cookie файлов. Токен находится по следующему пути:

.playerok.com | token

---

Структура библиотеки

PyPlayerokAPI/
│
├── account/
│   ├── Содержит все миксины для управления аккаунтом
│
├── models/
│   ├── Содержит все модели, описывающие аккаунт, сделки и т.д. (Pydantic)
│
├── stream/
|   ├── events/
|   |   ├── Содержит реализацию обработки всех типов ивентов
|   |
|   ├── listener/
|   |   ├── Содержит реализацию листенера и клиента websocket
|
├── types/
|   ├── Содержит константы
|
├── cacert.pem 
├── graphql.py Билдер payload запросов для GraphQL
├── transport.py Транспортный слой

---

Примеры использования

Предисловие

Стримы

AsyncPlayerokListener и AsyncMultiAccountListener запускают фоновых воркеров:

• WebSocket стрим (chatUpdated, chatMessageCreated, userUpdated)
• Deal стрим (ищет только новые созданные чаты с ивентом {{ITEM_PAID}})
• Review стрим (проверяет сделки которые ожидают новый отзыв)

Ивенты

Все события описываются в одной модели - PlayerokEvent. В мульти аккаунт моде обработчики получают:

• account: AccountClient
• event: PlayerokEvent или одну обертку:
• account_event: AccountEvent

Рекомендуемые сигнатуры хендлеров

async def handler(account: AccountClient, event: PlayerokEvent) -> None:
    ...

async def handler(account_event: AccountEvent) -> None:
    ...

Поддержка ивентов

Поддерживаются все типы EventTypes:

• CHAT_INITIALIZED - Чат инициализирован
• NEW_MESSAGE - Новое сообщение в чате
• NEW_DEAL - Создана новая сделка (когда покупатель оплатил товар)
• NEW_REVIEW - Новый отзыв от покупателя
• DEAL_CONFIRMED - Сделка подтверждена (покупатель подтвердил получение предмета)
• DEAL_CONFIRMED_AUTOMATICALLY - Сделка подтверждена автоматически (если покупатель долго не выходит на связь)
• DEAL_ROLLED_BACK - Продавец оформил возврат сделки
• DEAL_HAS_PROBLEM - Пользователь сообщил о проблеме в сделке
• DEAL_PROBLEM_RESOLVED - Проблема в сделке решена
• DEAL_STATUS_CHANGED - Статус сделки изменён
• ITEM_PAID - Пользователь оплатил предмет
• ITEM_SENT - Предмет отправлен (продавец подтвердил выполнение сделки)

Маппинг маркеров:

• {{ITEM_PAID}} -> NEW_DEAL, ITEM_PAID
• {{ITEM_SENT}} -> ITEM_SENT, DEAL_STATUS_CHANGED
• {{DEAL_CONFIRMED}} -> DEAL_CONFIRMED, DEAL_STATUS_CHANGED
• {{DEAL_CONFIRMED_AUTOMATICALLY}} -> DEAL_CONFIRMED_AUTOMATICALLY, DEAL_STATUS_CHANGED
• {{DEAL_ROLLED_BACK}} -> DEAL_ROLLED_BACK, DEAL_STATUS_CHANGED
• {{DEAL_HAS_PROBLEM}} -> DEAL_HAS_PROBLEM, DEAL_STATUS_CHANGED
• {{DEAL_PROBLEM_RESOLVED}} -> DEAL_PROBLEM_RESOLVED, DEAL_STATUS_CHANGED

Важные правила

1. Вызывать listener.start() только внутри запущенного loop.
2. Использовать полный JWT токен аккаунта.
3. Не использовать while True: asyncio.run(...).
4. Если используете декораторы, сохраняйте dispatch = True (дефолт).
5. Если хотите использовать очередь вручную, установите dispatch = False.

Так-же не забывайте, что

Используйте await asyncio.Event().wait() как бесконечный блокер когда:

1. Это простой standalone-скрипт
2. Все нужные воркеры уже запущены в фоне
3. Больше нет ни одного await, который держит loop живым

В ином другом случае, когда уже есть естественная точка блокировки - не используем. Пример:

• (aiogram) await dp.start_polling(bot)
• (ручное чтение очереди) while True: item = await listener.get()
• await server.serve_forever()
• и т.д.

---

Общие варианты использования

Стандартная инициализация аккаунта

from PyPlayerokAPI.account import AccountClient

account = AccountClient(
	token = "JWT_TOKEN_HERE", # обязательное поле
	user_agent = "", # необязательное поле (но желательное) (для разных аккаунтов используйте ранзые юзер агенты для избежания блокировки)
	proxy = "" # необязательное поле (но желательное)
)

print(account.account_data) # получаем информацию об аккаунте

Выставление предмета на продажу

from PyPlayerokAPI.account import AccountClient

account = AccountClient(token = "", user_agent = "")

# получаем информацию игры
game = account.get_game(slug = "minecraft")

# # получаем необходимую категорию (все доступные категории описаны в game.categories)
game_category = account.get_game_category(
    id = [category for category in game.categories if category.name == "Ключи"][0].id
)

# получаем варианты "доставки" товара в этой категории
game_obtaining_type_list = account.get_game_category_obtaining_types(
    game_category_id = game_category.id
)

# выбираем тип выдачи "Без входа в аккаунт"
choosed_obtaining_type = [
    obt_type for obt_type in game_obtaining_type_list.obtaining_types if obt_type.name == "Без входа в аккаунт"
][0]

# выбираем вариант версии игры (pc, mobile, ps, xbox)
gift_type_option = [
    gift_type for gift_type in game_category.options if gift_type.value == "pc"
]

# получаем поля с данными категории определенного типа выдачи
data_fields_list = account.get_game_category_data_fields(
    game_category_id = game_category.id,
    obtaining_type_id = choosed_obtaining_type.id
)

# Берем поле с данными о комментарии (все доступные поля описаны в data_fields_list.data_fields)
comment_data_field = [
    data_field for data_field in data_fields_list.data_fields if data_field.label == "Комментарий"
][0]

# Задаем значение данному полю, так как оно обязательное (не менее 10 симовлов)
comment_data_field.value = "Спасибо! Заказывайте у нас еще!"

# Берем поле с данными о ключе
key_data_field = [
    data_field for data_field in data_fields_list.data_fields if data_field.label == "Ключ"
][0]

# Задаем значение данному полю, так как оно обязательное
key_data_field.value = "J73D2-XXXXX-XXXXX-XXXXX-XXXXX"

# описываем путь к файлу фотографии для карточки товара
banner_attachment = "banner.jpg"

# вызываем метод создания товара
item = account.create_item(
    game_category_id = game_category.id, # указываем id категории игры
    obtaining_type_id = choosed_obtaining_type.id, # указывает id типа получения товара
    name = "Ключ MINECRAFT JAVA / BEDROCK", # указываем название товара
    price = 1700, # указываем цену товара
    description = """
🌹Активация через Веб-браузер:
1️⃣ Запустите веб-браузер и перейдите по адресу: https://redeem.microsoft.com
2️⃣ Войдите, используя свои учетные данные Microsoft
3️⃣Введите код активации и нажмите «Далее»; следуйте инструкциям для подтверждения
    """, # указываем описание товара
    options = gift_type_option, # указываем вариант товара
    data_fields = [comment_data_field, key_data_field], # указываем поля с данными предмета
    attachments = [banner_attachment] # указываем фотографии и тд
)

# Публикуем предмет
statuses = account.get_item_priority_statuses(
    item_id = item.id,
    item_price = str(item.price)
)

new_item = account.publish_item(
    item_id = item.id,
    priority_status_id = f"{next(status.id for status in statuses if status.price == 0 or status.name == "Обычный")}" # выставляем приоритет
)

---

Готовые варианты использования

Быстрый старт и простой одиночный клиент

import asyncio

from PyPlayerokAPI.account import AccountClient
from PyPlayerokAPI.stream import AsyncPlayerokListener
from PyPlayerokAPI.stream.events.account_event import AccountEvent
from PyPlayerokAPI.stream.events.event_wrapper.py import PlayerokEvent

TOKEN = ""
USER_AGENT = ""

async def main():
	account = AccountClient(token = TOKEN, user_agent = USER_AGENT)
	listener = AsyncPlayerokListener(account)
	router = listener.router
	
	@router.on_new_message()
	async def on_new_message(account: AccountClient, event: PlayerokEvent) -> None:
		text = event.message.text if event.message else None
		chat_id = event.chat.id if event.chat else None
		print("NEW_MESSAGE", chat_id, text)
	
	listener.start()
	await asyncio.Event().wait() 

if __name__ == "__main__":
	asyncio.run(main())

Множество клиентов

import asyncio

from PyPlayerokAPI.account import AccountClient
from PyPlayerokAPI.stream import AsyncMultiAccountListener
from PyPlayerokAPI.stream.events.event_wrapper import PlayerokEvent


ACCOUNTS = [
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",

    },
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    },
]

  
def account_label(account: AccountClient) -> str:
    return account.token[:10]

  
async def main():
    clients = [
        AccountClient(token = item["token"], user_agent = item["user_agent"])
        for item in ACCOUNTS
    ]

    listener = AsyncMultiAccountListener(clients)
    router = listener.router

    @router.on_new_message()
    async def on_new_message(account: AccountClient, event: PlayerokEvent):
        text = event.message.text if event.message else None
        print("[MSG]", account_label(account), text)

    @router.on_item_paid()
    async def on_item_paid(account: AccountClient, event: PlayerokEvent):
        deal_id = event.deal.id if event.deal else None
        print("[ITEM_PAID]", account_label(account), deal_id)

    listener.start()
    await asyncio.Event().wait()

  
if __name__ == "__main__":
    asyncio.run(main())

Мануал мод (чтение ивентов напрямую в обход диспатчера)

import asyncio

from PyPlayerokAPI.account import AccountClient
from PyPlayerokAPI.stream import AsyncMultiAccountListener
from PyPlayerokAPI.stream.events.account_event import AccountEvent

  
ACCOUNTS = [
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    },
]

  
async def main():
    clients = [
        AccountClient(token = item["token"], user_agent = item["user_agent"])
        for item in ACCOUNTS
    ]

    listener = AsyncMultiAccountListener(clients)

    # dispatch = False: без декораторов, ручной перебор очереди
    listener.start(dispatch = False)

    while True:
        item: AccountEvent = await listener.get()
        event = item.event
        print(
            "EVENT",
            item.account.token[:10],
            event.type.name,
            event.chat.id if event.chat else None,
        )

  
if __name__ == "__main__":
    asyncio.run(main())

Использование всех декораторов

import asyncio

from PyPlayerokAPI.account import AccountClient
from PyPlayerokAPI.stream import AsyncMultiAccountListener
from PyPlayerokAPI.stream.events.event_wrapper import PlayerokEvent

  
ACCOUNTS = [
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    },
]

  
def handle(event_name: str, account: AccountClient, event: PlayerokEvent):
    chat_id = event.chat.id if event.chat else None
    msg = event.message.text if event.message else None
    deal_id = event.deal.id if event.deal else None
    print(event_name, account.token[:10], chat_id, msg, deal_id)

  
async def main():
    clients = [
        AccountClient(token = item["token"], user_agent = item["user_agent"])
        for item in ACCOUNTS
    ]
    
    listener = AsyncMultiAccountListener(clients)
    router = listener.router

  
    @router.on_chat_initialized()
    async def on_chat_initialized(account: AccountClient, event: PlayerokEvent):
        handle("CHAT_INITIALIZED", account, event)

    @router.on_new_message()
    async def on_new_message(account: AccountClient, event: PlayerokEvent):
        handle("NEW_MESSAGE", account, event)

    @router.on_new_deal()
    async def on_new_deal(account: AccountClient, event: PlayerokEvent):
        handle("NEW_DEAL", account, event)

    @router.on_new_review()
    async def on_new_review(account: AccountClient, event: PlayerokEvent):
        handle("NEW_REVIEW", account, event)

    @router.on_deal_confirmed()
    async def on_deal_confirmed(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_CONFIRMED", account, event)

    @router.on_deal_confirmed_automatically()
    async def on_deal_confirmed_auto(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_CONFIRMED_AUTOMATICALLY", account, event)

    @router.on_deal_rolled_back()
    async def on_deal_rolled_back(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_ROLLED_BACK", account, event)

    @router.on_deal_has_problem()
    async def on_deal_has_problem(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_HAS_PROBLEM", account, event)

    @router.on_deal_problem_resolved()
    async def on_deal_problem_resolved(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_PROBLEM_RESOLVED", account, event)

    @router.on_deal_status_changed()
    async def on_deal_status_changed(account: AccountClient, event: PlayerokEvent):
        handle("DEAL_STATUS_CHANGED", account, event)

    @router.on_item_paid()
    async def on_item_paid(account: AccountClient, event: PlayerokEvent):
        handle("ITEM_PAID", account, event)

    @router.on_item_sent()
    async def on_item_sent(account: AccountClient, event: PlayerokEvent):
        handle("ITEM_SENT", account, event)

    listener.start()
    await asyncio.Event().wait()

  
if __name__ == "__main__":
    asyncio.run(main())

Использование в боте на основе aiogram

import asyncio

from aiogram import Bot, Dispatcher
from aiogram.filters import Command
from aiogram.types import Message

from PyPlayerokAPI.account import AccountClient
from PyPlayerokAPI.stream import AsyncMultiAccountListener
from PyPlayerokAPI.stream.events.event_wrapper import PlayerokEvent

  
BOT_TOKEN = "PUT_TELEGRAM_BOT_TOKEN_HERE"
ADMIN_CHAT_ID = 123456789

ACCOUNTS = [
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    },
    {
        "token": "",
        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    },
]

  
def build_clients() -> list[AccountClient]:
    return [
        AccountClient(token = item["token"], user_agent = item["user_agent"])
        for item in ACCOUNTS
    ]
  
  
def install_stream_handlers(listener: AsyncMultiAccountListener, bot: Bot):
    router = listener.router

    @router.on_new_message()
    async def on_new_message(account: AccountClient, event: PlayerokEvent):
        text = event.message.text if event.message and event.message.text else "<empty>"
        chat_id = event.chat.id if event.chat else "unknown"
        await bot.send_message(
            ADMIN_CHAT_ID,
            f"[{account.token[:10]}] NEW_MESSAGE chat={chat_id}\n{text}",
        )

    @router.on_item_paid()
    async def on_item_paid(account: AccountClient, event: PlayerokEvent):
        deal_id = event.deal.id if event.deal else "unknown"
        await bot.send_message(
            ADMIN_CHAT_ID,
            f"[{account.token[:10]}] ITEM_PAID deal={deal_id}",
        )

  
async def main():
    bot = Bot(token = BOT_TOKEN)
    dp = Dispatcher()

    @dp.message(Command("ping"))
    async def ping(message: Message):
        await message.answer("pong")

    listener = AsyncMultiAccountListener(build_clients())
    install_stream_handlers(listener, bot)
    listener.start()

    await dp.start_polling(bot)

  
if __name__ == "__main__":
    asyncio.run(main())

---

Дополнительная информация

Данная библиотека является переработанной и архитектурно переосмысленной версией проекта PlayerokAPI.

Кодовая база была полностью реорганизована с упором на:

• читаемость
• масштабируемость
• модульную архитектуру
• соответствие современным Python-практикам

Проект не является форком, а представляет собой самостоятельную реализацию с переработанной структурой.

---