抖音直播

一个 python 库，用于连接和读取 TikTok 的 LIVE 服务中的事件。

一个python库，通过连接到TikTok的内部网络广播推送服务，从TikTok的LIVE服务中实时接收和解码评论和礼物等直播事件。该库包含一个包装器，该包装器仅使用用户的连接到网络广播服务unique_id，并允许您加入您的直播以及其他流媒体的直播。使用 TikTokLive 不需要任何凭据。

该库是 由@zerodytrash 开发的 JavaScript TikTok-Live-Connector的 Python 实现，旨在为那些在 Python 中工作更舒服或需要其特定项目参数的用户提供替代方案。

加入支持不和谐并访问#support频道以获取问题、贡献和想法。随意提出缺少/新功能、修复等的拉取请求。

注意：这不是官方 API。这是一个逆向工程和研究项目。

更新：
由于 TikTok 的更改，v4.3.3之前的版本不再起作用。如果您使用的是不受支持的版本，请使用该pip install TikTokLive --upgrade命令升级到最新版本。

增加访问权限（API 密钥）

需要增加对 TikTok 服务器的访问权限？API 密钥可以通过在 Discord 上与我联系来获取。50% 的收益用于https://www.thetrevorproject.org/。

目录

主要信息

• 文档
• 贡献者
• 执照
• 快速入门示例

资源和指南

1. 大卫的介绍教程
2. 入门
3. 参数和选项
4. 客户端方法
5. 客户端属性
6. 抖音活动

TikTokLive 介绍教程

对于尝试入门的人，我不能推荐本教程。它简洁、信息丰富且易于理解，由Python TikTok-Api包的创建者David Teather创建。单击缩略图进行变形。

入门

1. 通过 pip 安装模块

pip install TikTokLive

2. 创建您的第一个聊天连接

from TikTokLive import TikTokLiveClient
from TikTokLive.types.events import CommentEvent, ConnectEvent

# Instantiate the client with the user's username
client: TikTokLiveClient = TikTokLiveClient(unique_id="@isaackogz")


# Define how you want to handle specific events via decorator
@client.on("connect")
async def on_connect(_: ConnectEvent):
    print("Connected to Room ID:", client.room_id)


# Notice no decorator?
async def on_comment(event: CommentEvent):
    print(f"{event.user.nickname} -> {event.comment}")


# Define handling an event via "callback"
client.add_listener("comment", on_comment)

if __name__ == '__main__':
    # Run the client and block the main thread
    # await client.start() to run non-blocking
    client.run()

有关更多示例，请参阅树中提供的示例文件夹。

参数和选项

要创建新TikTokLiveClient对象，需要以下参数。您可以选择通过 kwargs 添加配置选项。

TikTokLiveClient(unique_id, **options)

示例选项：

from TikTokLive import TikTokLiveClient

client: TikTokLiveClient = TikTokLiveClient(
    unique_id="@oldskoldj", **(
        {

            # Custom Asyncio event loop
            "loop": None,

            # Custom Client params
            "client_params": {},

            # Custom request headers
            "headers": {},

            # Custom timeout for Webcast API requests
            "timeout_ms": 1000,

            # How frequently to make requests the webcast API when long polling
            "ping_interval_ms": 1000,

            # Whether to process initial data (cached chats, etc.)
            "process_initial_data": True,

            # Whether to get extended gift info (Image URLs, etc.)
            "enable_extended_gift_info": True,

            # Whether to trust environment variables that provide proxies to be used in http requests
            "trust_env": False,

            # A dict object for proxies requests
            "proxies": {
                "http://": "http://username:password@localhost:8030",
                "https://": "http://420.69.420:8031",
            },

            # Set the language for Webcast responses (Changes extended_gift's language)
            "lang": "en-US",

            # Connect info (viewers, stream status, etc.)
            "fetch_room_info_on_connect": True,

            # Whether to allow Websocket connections
            "websocket_enabled": False,
            
            # Parameter to increase the amount of connections made per minute via a Sign Server API key. 
            # If you need this, contact the project maintainer.
            "sign_api_key": None

        }
    )
)

if __name__ == "__main__":
    client.run()

方法

一个TikTokLiveClient对象包含以下方法。

属性

活动

一个TikTokLiveClient对象具有以下事件。您可以通过执行client.add_listener("event_name", callable)或装饰@client.on("event_name")包含事件有效负载参数的函数来添加事件。

connect

当连接成功建立时触发。

@client.on("connect")
async def on_connect(event: ConnectEvent):
    print("Connected")

disconnect

当连接断开时触发。您可以致电start() 重新连接。请注意，您应该在尝试重新连接之前稍等片刻，以避免受到速率限制。

@client.on("disconnect")
async def on_disconnect(event: DisconnectEvent):
    print("Disconnected")

like

每次有人喜欢该流时触发。

@client.on("like")
async def on_like(event: LikeEvent):
    print("Someone liked the stream!")

join

每次有新人加入流时触发。

@client.on("join")
async def on_join(event: JoinEvent):
    print("Someone joined the stream!")

gift

每次收到礼物时触发。额外的信息可以从available_gifts客户端属性中看到。

注意：用户可以连续发送礼物。这会增加该data.gift.repeat_count值，直到用户终止连胜。在此期间，新的礼物事件被一次又一次地触发，data.gift.repeat_count价值增加。需要注意的是，在连胜结束后，会触发另一个礼物事件，通过data.gift.repeat_end:表示连胜结束1。这仅适用于带有data.gift.gift_type:的礼物1。这意味着即使用户只发送了一次gift_type:1礼物，您也会收到两次该事件。一次使用repeat_end:0一次使用repeat_end: 1。因此，该事件应按以下两种方式之一进行处理：

@client.on("gift")
async def on_gift(event: GiftEvent):
    # If it's type 1 and the streak is over
    if event.gift.gift_type == 1:
        if event.gift.repeat_end == 1:
            print(f"{event.user.uniqueId} sent {event.gift.repeat_count}x \"{event.gift.extended_gift.name}\"")

    # It's not type 1, which means it can't have a streak & is automatically over
    elif event.gift.gift_type != 1:
        print(f"{event.user.uniqueId} sent \"{event.gift.extended_gift.name}\"")

@client.on("gift")
async def on_gift(event: GiftEvent):
    # If it's type 1 and the streak is over
    if event.gift.streakable:
        if not event.gift.streaking:
            print(f"{event.user.uniqueId} sent {event.gift.repeat_count}x \"{event.gift.extended_gift.name}\"")

    # It's not type 1, which means it can't have a streak & is automatically over
    else:
        print(f"{event.user.uniqueId} sent \"{event.gift.extended_gift.name}\"")

follow

每次有人关注主播时触发。

@client.on("follow")
async def on_follow(event: FollowEvent):
    print("Someone followed the streamer!")

share

每次有人分享流时触发。

@client.on("share")
async def on_share(event: ShareEvent):
    print("Someone shared the streamer!")

viewer_count_update

每次更新观众人数时触发。默认情况下，此事件还会更新缓存的查看器计数。

@client.on("viewer_count_update")
async def on_connect(event: ViewerCountUpdateEvent):
    print("Received a new viewer count:", event.viewCount)

comment

每次有人评论直播时触发。

@client.on("comment")
async def on_connect(event: CommentEvent):
    print(f"{event.user.nickname} -> {event.comment}")

emote

当有人向实时聊天发送订阅表情评论时触发。

@client.on("emote")
async def on_connect(event: EmoteEvent):
    print(f"{event.user.nickname} -> {event.emote.image.imageUrl}")

envelope

当有人向 TikTok 流媒体发送信封（宝箱）时触发。

@client.on("envelope")
async def on_connect(event: EnvelopeEvent):
    print(f"{event.treasureBoxUser.uniqueId} -> {event.treasureBoxData}")

subscribe

当有人订阅 TikTok 流媒体时触发。

@client.on("subscribe")
async def on_connect(event: SubscribeEvent):
    print(f"{event.user.uniqueId} just subscribed to {client.unique_id}!")

weekly_ranking

发送每周排名更新时触发。

@client.on("weekly_ranking")
async def on_connect(event: WeeklyRankingEvent):
    print(f"{client.unique_id} is in the top {event.data.rankings.rank.id} streamers!")

mic_battle

麦克风对战开始时触发！

@client.on("mic_battle")
async def on_connect(event: MicBattleEvent):
    print(f"A Mic battle has started between {', '.join([user.battleGroup.user.uniqueId for user in event.battleUsers])}")

mic_armies

当收到有关麦克风战斗进展的信息时触发。

@client.on("mic_armies")
async def on_connect(event: MicArmiesEvent):
    print(f"Mic battle data: {event.battleUsers}")

more_share

当超过 5 或 10 个用户从查看者的共享链接加入时触发。

@client.on("more_share")
async def on_connect(event: MoreShareEvent):
    print(f"More than {event.amount} users have joined from {user.uniqueId}'s share link!")

live_end

当直播被主机终止时触发。

@client.on("live_end")
async def on_connect(event: LiveEndEvent):
    print(f"Livestream ended :(")

unknown

当收到此客户端尚未处理的未知事件时触发。

@client.on("unknown")
async def on_connect(event: UnknownEvent):
    print(event.as_dict, "<- This is my data as a dict!")

error

当客户端或错误处理程序中出现错误时触发。

如果代码中不存在此处理程序，则内部默认处理程序将在控制台中记录错误。如果添加了处理程序，则所有错误处理（包括日志记录）都取决于个人。

警告：如果您侦听错误事件并且不记录错误，那么您将看不到错误发生的时间。这是因为侦听错误事件会导致默认事件被覆盖/关闭。

@client.on("error")
async def on_connect(error: Exception):
    # Handle the error
    if isinstance(error, SomeRandomError):
        print("Handle Some Error")
        return

    # Otherwise, log the error
    client._log_error(error)

贡献者

• Isaac Kogan -初始工作和主要维护者- isaackogan
• Zerody -逆向工程与支持- Zerody
• Davincible -逆向工程流下载 - davincible
• David Teather - TikTokLive 介绍教程- davidteather

另请参阅参与此项目的贡献者的完整列表。

执照

这个项目是在 MIT 许可下获得许可的 - 有关详细信息，请参阅LICENSE文件。