оригинальная статья на GitHub - TikTokLive
Что такое TikTokLive?
TikTokLive это библиотека написанная на Python которая даёт вам возможность читать события с стримов в приложении TikTok.
Данная библиотека это перенос библиотеки TikTokLive-Connector с JavaScript на Python.
И так как же использовать TikTokLive\
- Первым делом устанавливаем саму библиотеку.
pip install TikTokLive
- Создаём соединением с чатом.
from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent, ConnectEvent
# К какому пользователю будем подключаться.
client: TikTokLiveClient = TikTokLiveClient(unique_id="@monzik.neofc")
# Обработка определенных событий с помощью декоратора.
@client.on("connect")
async def on_connect(_: ConnectEvent):
print("Подключился к комнате:", client.room_id)
# В случае если нет декоратора
async def on_comment(event: CommentEvent):
print(f"{event.user.nickname} -> {event.comment}")
# Определить обработку события через «обратный вызов».
client.add_listener("comment", on_comment)
if __name__ == '__main__':
# Запуск клиента и блокировка основного потока.
# await client.start() to run non-blocking
client.run()
Параметры и опции:
unique_id - Обязательный параметр.
**Уникальное имя пользователя, таковыми являются URL или NickName - https://www.tiktok.com/@monzik.neofc => monzik.neofc**.
debug - Необязательный параметр.
Включает "отладку" для необработанных данных.
**options - Необязательный параметр.
Здесь вы можете установить следующие необязательные свойства подключения. Если вы не укажете значение, будет использоваться значение по умолчанию.
process_initial_data (по умолчанию: True)
Укажите, хотите ли вы обрабатывать исходные данные, которые включают в себя старые сообщения за последние секунды.
fetch_room_info_on_connect (по умолчанию: True)
Хотите ли вы получать всю информацию о комнате при запуске. Если эта опция включена, подключение к автономным комнатам будет запрещено. Если включено, результат подключения будет содержать информацию о комнате через атрибут room_info. Вы также можете вручную получить информацию о комнате (даже в не подключённом состоянии), используя метод retrieve_room_info().
enable_extended_gift_info (по умолчанию: True)
Укажите, хотите ли вы получать расширенную информацию о подарках, такую как название подарка, стоимость и изображения, которые вы можете получить с помощью атрибута available_gifts.
ping_interval_ms (по умолчанию: 1000)
Как часто запросы к API веб-трансляции будут отправляться.
client_params (по умолчанию: {})
Пользовательские параметры клиента для Webcast API.
client_params (по умолчанию: {})
Пользовательские заголовки запросов передаются в aiohttp.
timeout_ms (по умолчанию: 1000)
Как долго ждать, прежде чем запрос должен завершиться ошибкой.
loop (по умолчанию: None)
При желании укажите собственный цикл обработки событий asyncio для использования клиентом. Если установлено значение None, клиент извлекает текущий активный цикл или создаёт новый. Эта опция в основном полезна для людей, пытающихся вложить asyncio.
trust_env (по умолчанию: False)
Следует ли доверять переменным среды, которые предоставляют прокси-серверы в запросах httpx.
proxies (по умолчанию: False)
При включение данного параметра TikTokLive будет использовать прокси-запросы включив переадресацию для аргумента «Proxy» HTTPX. Соединения через веб-сокет НЕ будут проксироваться.
lang (по умолчанию: en)
Измените язык. Полезная нагрузка будет на английском языке, но интерфейсный контент будет на желаемом языке!
websocket_enabled (по умолчанию: True)
Использовать ли веб-сокеты или полагаться на исключительно длинный опрос.
sign_api_key (по умолчанию: None)
Параметр для увеличения количества разрешенных подключений в минуту с помощью ключа API Sign Server.
Обычные функции:
from TikTokLive import TikTokLiveClient
client: TikTokLiveClient = TikTokLiveClient(
unique_id="@oldskoldj", **(
{
# Кастомный Asyncio цикл события
"loop": None,
# Кастомный Client параметр
"client_params": {},
# Пользовательские заголовки запросов для включения в HTTP-запросы
"request_headers": {},
# Пользовательские заголовки веб-сокетов для включения при подключении к TikTok WebSocket
"websocket_headers": {},
# Пользовательский тайм-аут для запросов API веб-трансляции
"timeout_ms": 1000,
# Как делать частые запросы к API веб-трансляции при длительном опросе
"ping_interval_ms": 1000,
# Обрабатывать ли исходные данные (Кэшированные чаты)
"process_initial_data": True,
# Получать ли расширенную информацию о подарке (URLs картинки)
"enable_extended_gift_info": True,
# Доверять ли переменным среды, которые предоставляют прокси для использования в HTTP-запросах
"trust_env": False,
# Объект dict для запросов прокси
"proxies": {
"http://": "http://username:password@localhost:8030",
"https://": "http://420.69.420:8031",
},
# Установите языков для веб-трансляции (Изменяет язык extended_gift)
"lang": "en-US",
# Информация о подключении (Зрители, Статус трансляции)
"fetch_room_info_on_connect": True,
# Разрешать ли подключения через Websocket
"websocket_enabled": False,
# Параметр для увеличения количества подключений в минуту с помощью ключа API Sign Server.
"sign_api_key": None
}
)
)
if __name__ == "__main__":
client.run()
Методы
Объект TikTokLiveClient содержит следующие методы.
run - Начинает подключение к живому чату, блокируя основной поток (синхронизация).start - Подключается к живому чату, не блокируя основной поток (асинхронно).stop - Отключает подключение к живому чату.retrieve_room_info - Получает текущую информацию о комнате из TikTok API.retrieve_available_gifts - Получает список доступных подарков для комнаты и добавляет его к атрибуту extended_gift объекта Gift в событии подарка, если он включен.add_listener - Добавляет асинхронную функцию слушателя (или вы можете украсить функцию @client.on()) и принимает два параметра, имя события и полезную нагрузку, AbstractEvent.download - Начать загрузку видео в прямом эфире в течение заданной продолжительности или до остановки с помощью метода stop_download.stop_download - Остановите загрузку видео в прямом эфире, если оно загружается в данный момент, иначе выдаётся ошибка.send_message - Отправьте сообщение в чат. Это реализовано лишь частично. См. пример ниже для получения дополнительной информации.
Пример:
from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent
client = TikTokLiveClient("@monzik.neofc")
async def sign_url(raw_url: str, session_id: str):
"""
Вам нужно будет создать свою СОБСТВЕННУЮ функцию, чтобы изменить HTTP-запрос по своему вкусу, чтобы он проходил аутентификацию TikTok.
TikTokLive не может и не будет предоставлять подписи, но если вам нужна эта функциональность, она здесь.
:param raw_url: URL-адрес, требующий подписи
:param session_id: Идентификатор сеанса, отправляющий сообщение
:return: None
"""
# Вам нужно будет сгенерировать каждый из
signed_url: str = raw_url + (
f"&msToken={'СГЕНЕРИРУЙ МЕНЯ'}"
f"&X-Bogus={'СГЕНЕРИРУЙ МЕНЯ'}"
f"&User-Agent={'СГЕНЕРИРУЙ МЕНЯ'}"
f"&browserVersion={'СГЕНЕРИРУЙ МЕНЯ'}"
f"&browserName={'СГЕНЕРИРУЙ МЕНЯ'}"
f"&_signature={'СГЕНЕРИРУЙ МЕНЯ}"
)
# Вам нужно будет предоставить свои собственные заголовки
headers: dict = {
"Cookie": "ttwid=СГЕНЕРИРУЙ_TTWID;",
**client._http.headers
}
return signed_url, headers
@client.on("message")
async def on_ping(event: CommentEvent):
"""
Когда кто-то запускает команду /ping, выберите, как реагировать
:param event: Комментарий к событию
:return: None
"""
# If not ping, return
if event.comment.lower() != "/ping":
return
# Reply with Pong
reply: str = f"{event.user.uniqueId} Pong!"
print(f"Бот ответит в чате с \"{reply}\"")
await client.send_message(
text=reply,
sign_url_fn=sign_url,
session_id="АЙДИ_СЕССИИ_ТУТ"
)
if __name__ == '__main__':
"""
Пример показывающий как вы можете отправлять комментарии в свою прямую трансляцию.
"""
client.run()
Атрибуты
viewer_count - Количество людей, которые в данный момент смотрят прямую трансляцию.room_id - Идентификатор комнаты прямой трансляции, к которой в данный момент подключен клиент.room_info - Информация о данной комнате прямой трансляции.unique_id - Имя пользователя человека в TikTok, к которому в данный момент подключен клиент.connected - Подключен ли клиент в настоящее время к прямой трансляцииavailable_gifts - Словарь, содержащий пары K:V Dict[int, ExtendedGift]proxies - Получите текущие прокси, используемые для HTTP-запросов. Примечание. Чтобы установить активные прокси, установите значение этого атрибута.
События
Объект TikTokLiveClient имеет следующие события. Вы можете добавить события, выполнив client.add_listener("event_name", callable) или украсив функцию @client.on("event_name"), которая включает параметр полезной нагрузки события.
Соединено
Запускается, когда соединение успешно установлено.
@client.on("connect")
async def on_connect(event: ConnectEvent):
print("Подключено")
Отключено
Срабатывает при разрыве соединения. Вы можете вызвать start() для повторного подключения. Обратите внимание, что вам следует немного подождать, прежде чем пытаться переподключаться, чтобы избежать ограничения скорости.
@client.on("disconnect")
async def on_disconnect(event: DisconnectEvent):
print("Отключено")
Лайк
Запускается каждый раз, когда кому-то лайкнул трансляцию.
@client.on("like")
async def on_like(event: LikeEvent):
print("Кто-то лайкнул стрим!")
Присоединился
Запускается каждый раз, когда новый человек присоединяется к Трансляции.
@client.on("join")
async def on_join(event: JoinEvent):
print("Кто-то подключился к трансляции!")
Подарок
Срабатывает каждый раз, когда приходит подарок. Дополнительную информацию можно получить из клиентского атрибута available_gifts.
@client.on("gift")
async def on_gift(event: GiftEvent):
# Если это тип 1 и полоса закончилась
if event.gift.gift_type == 1:
if event.gift.repeat_end == 1:
print(f"{event.user.uniqueId} Отправил {event.gift.repeat_count}x \"{event.gift.extended_gift.name}\"")
# Это не тип 1, что означает, что он не может иметь полосу и автоматически заканчивается.
elif event.gift.gift_type != 1:
print(f"{event.user.uniqueId} Отправил \"{event.gift.extended_gift.name}\"")
@client.on("gift")
async def on_gift(event: GiftEvent):
# Если это тип 1 и полоса закончилась
if event.gift.streakable:
if not event.gift.streaking:
print(f"{event.user.uniqueId} Отправил {event.gift.repeat_count}x \"{event.gift.extended_gift.name}\"")
# Это не тип 1, что означает, что он не может иметь полосу и автоматически заканчивается.
else:
print(f"{event.user.uniqueId} Отправил \"{event.gift.extended_gift.name}\"")
Подписался
Запускается каждый раз, когда кто-то подписался на стримера.
@client.on("follow")
async def on_follow(event: FollowEvent):
print("Кто-то подписался на стримера!")
Поделился
Запускается каждый раз, когда кто-то делится трансляцией.
@client.on("share")
async def on_share(event: ShareEvent):
print("Кто-то поделился трансляцией!")
Обновление счётчика зрителей
Запускается каждый раз, когда обновляется количество зрителей. Это событие также по умолчанию обновляет кэшированное количество зрителей.
@client.on("viewer_count_update")
async def on_connect(event: ViewerCountUpdateEvent):
print("Текущее количество зрителей:", event.viewCount)
Комментарий
Запускается каждый раз, когда кто-то комментирует прямую трансляцию.
@client.on("comment")
async def on_connect(event: CommentEvent):
print(f"{event.user.nickname} -> {event.comment}")
Эмоции
Запускается, когда кто-то эмодзи подписки в чат.
@client.on("emote")
async def on_connect(event: EmoteEvent):
print(f"{event.user.nickname} -> {event.emote.image.imageUrl}")
Конверт
Срабатывает, когда кто-то отправляет конверт (сундук с сокровищами) стримеру TikTok.
@client.on("envelope")
async def on_connect(event: EnvelopeEvent):
print(f"{event.treasureBoxUser.uniqueId} -> {event.treasureBoxData}")
Еженедельный рейтинг
Запускается при отправке еженедельного обновления рейтинга.
@client.on("weekly_ranking")
async def on_connect(event: WeeklyRankingEvent):
print(f"{client.unique_id} В топе на {event.data.rankings.rank.id} месте стримеров!")
Батлы
Срабатывает, когда начинается батл!
@client.on("mic_battle")
async def on_connect(event: MicBattleEvent):
print(f"Начался батл между {', '.join([user.battleGroup.user.uniqueId for user in event.battleUsers])}")
Информация об исходе батла
Срабатывает при получении информации о ходе батла.
@client.on("mic_armies")
async def on_connect(event: MicArmiesEvent):
print(f"Исход батла: {event.battleUsers}")
Сразу несколько человек подключились по приглашению трансляцией
Запускается, когда более 5 или 10 пользователей присоединяются по ссылке общего доступа зрителя.
@client.on("more_share")
async def on_connect(event: MoreShareEvent):
print(f"Более {event.amount} Пользователи подключились по ссылке приглашения от {user.uniqueId}!")
Конец трансляции
Запускается, когда трансляция прерывается хостом.
@client.on("live_end")
async def on_connect(event: LiveEndEvent):
print(f"Трансляция закончена :(")
Неизвестно
Запускается при получении неизвестного события, которое ещё не обработано этим клиентом.
@client.on("unknown")
async def on_connect(event: UnknownEvent):
print(event.as_dict, "<- Данное событие не найдено!")
Ошибка
Запускается при возникновении ошибки в клиенте или оброботчиках ошибок.
Если этот обработчик отсутствует в коде, внутренний обработчик по умолчанию будет регистрировать ошибки в консоли. Если добавлен обработчик, вся обработка ошибок (включая ведение журнала) остаётся на усмотрение пользователя.
Предупреждение: Если вы прослушиваете событие ошибки и не регистрируете ошибки, вы не увидите, когда произойдет ошибка. Это связано с тем, что прослушивание события ошибки приводит к переопределению/отключению события по умолчанию.
@client.on("error")
async def on_connect(error: Exception):
# Обработать ошибку
if isinstance(error, SomeRandomError):
print("Произошла какая-то ошибка")
return
# В противном случае запишите ошибку
client._log_error(error)
Данную статью я переводил 2 с половиной часа, так же хотел бы сказать то что благодаря этой библиотеки можно создать АВТО-СТРИМЫ такие как мини игры в которых за донат будет меняться видео или же будут убывать hp у монстра или то что придет вам в голову.
SourceCode
