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

def main()
    chat = LiveChat("G1w62uEMZ74", callback = func)
    while chat.is_alive():
      time.sleep(3)
      #other background operation.
   
#callback function is automatically called periodically.
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():
      await asyncio.sleep(3)
      #other background operation.

#callback function is automatically called periodically.
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)
        #other background operation here.

#callback function is automatically called periodically.
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.elapsedTime}> [{c.author.name}]-{c.message} {c.amountString}")
        await asyncio.sleep(tick/1000)

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

Structure of Default Processor

Each item can be got with items() function.

name	type	remarks
type	str	"superChat","textMessage","superSticker","newSponsor"
id	str
message	str	emojis are represented by ":(shortcut text):"
messageEx	str	list of message texts and emoji URLs.
timestamp	int	unixtime milliseconds
datetime	str	ex. "2019-10-10 12:34:56"
elapsedTime	str	elapsed time. (ex. "1:02:27") *Replay Only.
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	*chatter's channel ID. NOT broadcasting video's channel ID.
channelUrl	str
imageUrl	str
badgeUrl	str
isVerified	bool
isChatOwner	bool
isChatSponsor	bool
isChatModerator	bool

Licence

Author

taizan-hokuto

twitter:@taizan205

5.3 KiB Raw Blame History