Connect kick chat

ConnectKickChatPusher.py

# TeletextChat - Kick.com Chat Client
# Project: TeletextKickChat
# GitHub: devozdemirhasancan
# Disclaimer: This project is not affiliated with Kick.com in any way.

import json
import websockets
import tls_client
import sys

class KickChat:
    def __init__(self, channel_name):
        self.channel_name = channel_name
        self.ws_url = "wss://ws-us2.pusher.com/app/32cbd69e4b950bf97679?protocol=7&client=js&version=7.6.0&flash=false"
        self.chatroom_id = None
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept": "application/json",
            "Referer": f"https://kick.com/{channel_name}"
        }

    def get_channel_info(self):
        """Fetch channel information from Kick.com API with tls_client"""
        try:
            session = tls_client.Session(
                client_identifier="chrome_120",
                random_tls_extension_order=True
            )
            response = session.get(
                f"https://kick.com/api/v2/channels/{self.channel_name}",
                headers=self.headers
            )
            if response.status_code == 200:
                data = response.json()
                self.chatroom_id = data.get('chatroom', {}).get('id')
                print(f"Channel Info - Chatroom ID: {self.chatroom_id}")
                return True
            else:
                print(f"Error fetching channel info: Status Code {response.status_code} - {response.text[:200]}...")
                return False
        except Exception as e:
            print(f"Error fetching channel info: {e}")
            return False

    def connect(self):
        """Connect to Pusher WebSocket and listen for chat messages (sync version)"""
        import asyncio
        asyncio.run(self._async_connect())

    async def _async_connect(self):
        """Internal async connection method"""
        if not self.chatroom_id:
            if not self.get_channel_info():
                print("Failed to get channel info.")
                return

        print("Connecting to Pusher...")
        try:
            async with websockets.connect(self.ws_url) as websocket:
                # Receive connection response
                connection_response = await websocket.recv()
                connection_data = json.loads(connection_response if isinstance(connection_response, str) else connection_response.decode())
                socket_id_data = json.loads(connection_data.get("data", "{}") if isinstance(connection_data.get("data"), str) else connection_data.get("data", "{}"))
                socket_id = socket_id_data.get("socket_id", "Unknown")
                print(f"Connected with socket ID: {socket_id}")

                # Subscribe to the chatroom
                subscribe_msg = {
                    "event": "pusher:subscribe",
                    "data": {
                        "auth": "",
                        "channel": f"chatrooms.{self.chatroom_id}.v2"
                    }
                }
                await websocket.send(json.dumps(subscribe_msg))
                join_response = await websocket.recv()
                join_data = json.loads(join_response if isinstance(join_response, str) else join_response.decode())
                print(f"Subscribed to chatroom: {join_data.get('event', 'Unknown event')}")
                print(f"Connected to Kick channel '{self.channel_name}'")

                # Listen for messages
                while True:
                    try:
                        message = await websocket.recv()
                        msg_data = json.loads(message if isinstance(message, str) else message.decode())
                        if msg_data.get("event") == "App\\Events\\ChatMessageEvent":
                            data_str = msg_data.get("data", "{}")
                            chat_data = json.loads(data_str if isinstance(data_str, str) else data_str.decode()) if data_str else {}
                            sender = chat_data.get("sender", {})
                            username = sender.get("username", "Unknown")
                            content = chat_data.get("content", "")
                            print(f"[Chat] {username}: {content}")
                    except Exception as e:
                        print(f"Error processing message: {e}")
                        continue
        except Exception as e:
            print(f"WebSocket connection error: {e}")

if __name__ == "__main__":
    # Get channel name from command line argument or prompt user
    if len(sys.argv) > 1:
        channel_name = sys.argv[1]
    else:
        channel_name = input("Enter Kick.com channel name: ")

    chat = KickChat(channel_name)
    try:
        chat.connect()
    except Exception as e:
        print(f"An error occurred: {e}") 

juj

juj

Wise words