pytchat

pytchat is a python library for fetching youtube live chat.

Description

pytchat is a python library for fetching youtube live chat without using youtube api, Selenium or BeautifulSoup.

Other features:

Customizable chat data processors including yt api compatible one.
Available on asyncio context.
Quick fetching of initial chat data by generating continuation params instead of web scraping.

For more detailed information, see wiki.

Install

pip install pytchat

Demo

Examples

on-demand mode

from pytchat import LiveChat

chat = LiveChat("G1w62uEMZ74")
while chat.is_alive():
    data = chat.get()
    for c in data.items:
        print(f"{c.datetime} [{c.author.name}]-{c.message} {c.amountString}")
        data.tick()

callback mode

from pytchat import LiveChat
import time

chat = LiveChat("G1w62uEMZ74", callback = func)
while chat.is_alive():
    #other background operation here.
    time.sleep(3)

def func(data):
    for c in data.items:
        print(f"{c.datetime} [{c.author.name}]-{c.message} {c.amountString}")
        data.tick()

asyncio context:

from pytchat import LiveChatAsync
import asyncio

async def main():
    chat = LiveChatAsync("G1w62uEMZ74", callback = func)
    while chat.is_alive():
        #other background operation here.
        await asyncio.sleep(3)

async def func(data):
    for c in data.items:
        print(f"{c.datetime} [{c.author.name}]-{c.message} {c.amountString}")
        await data.tick_async()

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

youtube api compatible processor:

from pytchat import LiveChat, CompatibleProcessor

chat = LiveChat("G1w62uEMZ74", 
     processor = CompatibleProcessor() )

while chat.is_alive():
    data = chat.get()
    polling = data["pollingIntervalMillis"]/1000
    for c in data["items"]:
        if c.get("snippet"):
            print(f"[{c['authorDetails']['displayName']}]"
                    f"-{c['snippet']['displayMessage']}")
            time.sleep(polling/len(data["items"]))

replay:

from pytchat import ReplayChatAsync
import asyncio

async def main():
    chat = ReplayChatAsync("G1w62uEMZ74", seektime = 1000, callback = func)
    while chat.is_alive():
        #other background operation here.
        await asyncio.sleep(3)

async def func(data):
    for count in range(0,len(data.items)):
        c= data.items[count]
        if count!=len(data.items):
            tick=data.items[count+1].timestamp -data.items[count].timestamp
        else:
            tick=0
        print(f"<{c.timestampText}> [{c.author.name}]-{c.message} {c.amountString}")
        await asyncio.sleep(tick/1000)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Chatdata Structure of Default Processor

Structure of each item which got from items() function.

name	type	remarks
type	str	"superChat","textMessage","superSticker","newSponsor"
id	str
message	str	emojis are represented by ":(shortcut text):"
timestamp	int	unixtime milliseconds
datetime	str	ex. "2019-10-10 12:34:56"
timestampText	str	elapsed time. (ex. "1:02:27")
amountValue	float	ex. 1,234.0
amountString	str	ex. "$ 1,234"
currency	str	ISO 4217 currency codes (ex. "USD")
bgColor	int	RGB Int
author	object	see below

Structure of author object.

name	type	remarks
name	str
channelId	str
channelUrl	str
imageUrl	str
badgeUrl	str
isVerified	bool
isChatOwner	bool
isChatSponsor	bool
isChatModerator	bool

Licence

Author

taizan-hokuto

twitter:@taizan205

5.0 KiB Raw Blame History