Skip to content

Instantly share code, notes, and snippets.

@ouor

ouor/chzzk.py

Created November 21, 2025 16:21
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save ouor/542080aaef344e31185d976b02926244 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save ouor/542080aaef344e31185d976b02926244 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
chzzk.py
import json
import datetime
import logging
import requests # pip install requests
from typing import Literal
from websocket import WebSocket # pip install websocket-client
from http.cookiejar import MozillaCookieJar
from dataclasses import dataclass
# -------------------------------------
# 1. 명령어 타입 정의 및 공통 요청 헤더
# -------------------------------------
CHZZK_CHAT_CMD = {
'ping' : 0,
'pong' : 10000,
'connect' : 100,
'send_chat' : 3101,
'request_recent_chat' : 5101,
'chat' : 93101,
'donation' : 93102,
}
HEADERS = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/135.0.0.0 Safari/537.36'}
# ---------------------
# 2. API 호출 함수들
# ---------------------
def fetch_chatChannelId(streamer: str, cookies: dict) -> str:
url = f'https://api.chzzk.naver.com/polling/v2/channels/{streamer}/live-status'
response = requests.get(url, cookies=cookies, headers=HEADERS)
response.raise_for_status()
return response.json()['content']['chatChannelId']
def fetch_channelName(streamer: str) -> str:
url = f'https://api.chzzk.naver.com/service/v1/channels/{streamer}'
response = requests.get(url, headers=HEADERS)
response.raise_for_status()
return response.json()['content']['channelName']
def fetch_accessToken(chatChannelId: str, cookies: dict):
url = f'https://comm-api.game.naver.com/nng_main/v1/chats/access-token?channelId={chatChannelId}&chatType=STREAMING'
response = requests.get(url, cookies=cookies, headers=HEADERS)
response.raise_for_status()
data = response.json()['content']
return data['accessToken'], data['extraToken']
def fetch_userIdHash(cookies: dict) -> str:
url = 'https://comm-api.game.naver.com/nng_main/v1/user/getUserStatus'
response = requests.get(url, cookies=cookies, headers=HEADERS)
response.raise_for_status()
return response.json()['content']['userIdHash']
# ---------------------
# 3. 채팅 클래스
# ---------------------
@dataclass
class Chat:
time: str
type: Literal["채팅", "후원"]
nick: str
msg: str
class ChzzkChat:
"""
ChzzkChat 클래스를 직접 사용하여:
- 외부 모듈에서 생성 후 .listen_messages()를 통해 제너레이터 형태로 메시지 수신
"""
def __init__(
self,
streamer_id: str,
cookie_file: str = 'cookies.txt',
log_file: str = '',
log_stream: bool = False
):
"""
:param streamer_id: 스트리머 ID
:param cookie_file: 네이버 로그인에 사용되는 쿠키 (Netscape format)
:param log_file: 로그 파일 경로 (없으면 콘솔 출력만)
:param log_stream: 콘솔 로그 출력 여부
"""
self.streamer_id = streamer_id
self.cookies = {}
cookies = MozillaCookieJar(cookie_file)
cookies.load()
for c in cookies:
if c.name in ['NID_AUT', 'NID_SES']:
self.cookies[c.name] = c.value
self.logger = self._get_default_logger(log_file, log_stream)
# 웹소켓 URL
self.websocket_url = "wss://kr-ss1.chat.naver.com/chat"
# 최근 메시지 몇 개 불러올지
self.recent_message_count = 50
# 초기 정보 세팅
self.userIdHash = fetch_userIdHash(self.cookies)
self.chatChannelId = fetch_chatChannelId(self.streamer_id, self.cookies)
self.channelName = fetch_channelName(self.streamer_id)
self.accessToken, self.extraToken = fetch_accessToken(self.chatChannelId, self.cookies)
# 웹소켓 연결 (connect)
self.sock = None
self.sid = None
self.connect()
def _get_default_logger(self, log_file: str, log_stream: bool) -> logging.Logger:
"""
내부에서 사용할 기본 로거를 설정하는 함수
:param log_file: 로그 파일의 경로
:param log_stream: 로그를 출력할지 여부
"""
formatter = logging.Formatter('%(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
# 파일 핸들러
if log_file:
file_handler = logging.FileHandler(log_file, mode='w', encoding='utf-8')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
# 콘솔 핸들러
if log_stream:
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)
return logger
def connect(self):
"""
웹소켓에 연결하여 sid를 받아오고, 최근 채팅 메시지도 요청
"""
try:
# 매번 chatChannelId, accessToken, extraToken 갱신
self.chatChannelId = fetch_chatChannelId(self.streamer_id, self.cookies)
self.accessToken, self.extraToken = fetch_accessToken(self.chatChannelId, self.cookies)
if self.sock:
self.sock.close()
sock = WebSocket()
sock.connect(self.websocket_url)
self.logger.info(f"[INFO] {self.channelName} 채팅창에 연결 시도 중...")
# 공통 dict
default_dict = {
"ver": "2",
"svcid": "game",
"cid": self.chatChannelId,
}
# connect 요청
connect_dict = {
"cmd": CHZZK_CHAT_CMD['connect'],
"tid": 1,
"bdy": {
"uid": self.userIdHash,
"devType": 2001,
"accTkn": self.accessToken,
"auth": "SEND"
}
}
sock.send(json.dumps({**default_dict, **connect_dict}))
resp = json.loads(sock.recv())
# sid 획득
self.sid = resp['bdy']['sid']
# 최근 채팅 요청
recent_dict = {
"cmd": CHZZK_CHAT_CMD['request_recent_chat'],
"tid": 2,
"sid": self.sid,
"bdy": {
"recentMessageCount": self.recent_message_count
}
}
sock.send(json.dumps({**default_dict, **recent_dict}))
# 최근 메시지 응답
sock.recv() # 굳이 파싱하지 않아도 되므로 버림
self.sock = sock
self.logger.info(f"[INFO] {self.channelName} 채팅창 연결 완료 (SID: {self.sid})")
except Exception as e:
self.logger.error(f"[ERROR] 소켓 연결 실패: {e}")
self.sock = None
def send(self, message: str):
"""
채팅 메시지 전송 메서드
"""
if not self.sock:
self.logger.warning("[WARN] 소켓 연결되어 있지 않아 메시지 전송 불가!")
return
default_dict = {
"ver": 2,
"svcid": "game",
"cid": self.chatChannelId,
}
extras = {
"chatType": "STREAMING",
"emojis": "",
"osType": "PC",
"extraToken": self.extraToken,
"streamingChannelId": self.chatChannelId
}
send_dict = {
"tid": 3,
"cmd": CHZZK_CHAT_CMD['send_chat'],
"retry": False,
"sid": self.sid,
"bdy": {
"msg": message,
"msgTypeCode": 1,
"extras": json.dumps(extras),
"msgTime": int(datetime.datetime.now().timestamp())
}
}
try:
self.sock.send(json.dumps({**default_dict, **send_dict}))
except Exception as e:
self.logger.error(f"[ERROR] 메시지 전송 실패: {e}")
# 필요 시 재연결 로직을 추가할 수도 있음
def listen_messages(self):
"""
웹소켓에서 수신되는 메시지를 제너레이터로 yield합니다.
외부에서 for 루프로 호출해 사용 가능.
yield되는 값 예시:
{
"time": "2025-04-07 15:24:10",
"type": "채팅" or "후원",
"nickname": "닉네임",
"message": "메시지 내용"
}
"""
while True:
try:
raw_message = self.sock.recv() # 블로킹
except KeyboardInterrupt:
self.logger.info("[INFO] 사용자가 종료를 요청했습니다.")
break
except Exception as e:
self.logger.error(f"[ERROR] 소켓 recv 중 문제 발생: {e}, 재연결 시도...")
self.connect()
continue
try:
msg_json = json.loads(raw_message)
except json.JSONDecodeError as e:
self.logger.error(f"[ERROR] 메시지 JSON 디코딩 실패: {e}")
continue
chat_cmd = msg_json.get('cmd')
if chat_cmd is None:
continue
# ping/pong 처리
if chat_cmd == CHZZK_CHAT_CMD['ping']:
# pong 응답
pong_dict = {"ver": "2", "cmd": CHZZK_CHAT_CMD['pong']}
self.sock.send(json.dumps(pong_dict))
# 방송 재시작 등으로 chatChannelId 변동 시 재연결
try:
updated_id = fetch_chatChannelId(self.streamer_id, self.cookies)
if self.chatChannelId != updated_id:
self.logger.info("[INFO] 방송 재시작 감지, chatChannelId 변경으로 재연결")
self.connect()
except Exception as e:
self.logger.error(f"[ERROR] chatChannelId 재확인 실패: {e}")
continue
# 채팅/후원 메시지 처리
if chat_cmd in (CHZZK_CHAT_CMD['chat'], CHZZK_CHAT_CMD['donation']):
chat_type = '채팅' if chat_cmd == CHZZK_CHAT_CMD['chat'] else '후원'
bdy_list = msg_json.get('bdy', [])
for chat_data in bdy_list:
# 보낸이 식별
if chat_data.get('uid') == 'anonymous':
nickname = '익명의 후원자'
else:
try:
profile_data = json.loads(chat_data.get('profile', '{}'))
nickname = profile_data.get("nickname", "알 수 없음")
except Exception:
nickname = "알 수 없음"
if 'msg' not in chat_data:
continue
chat_msg = chat_data["msg"]
# 시간 정보
now = datetime.datetime.fromtimestamp(chat_data['msgTime'] / 1000)
str_time = now.strftime('%Y-%m-%d %H:%M:%S')
self.logger.info(f"{str_time}|{chat_type}|{nickname}|{chat_msg}")
chat = Chat(str_time, chat_type, nickname, chat_msg)
yield chat
if __name__ == '__main__':
chat_client = ChzzkChat('1906dd57f578c255feca54700bcccfc9', log_file='log.txt')
for chat in chat_client.listen_messages():
if chat.type == "채팅":
print(chat.nick + ': ' + chat.msg)
else:
print("---\n" + chat.nick + ': ' + chat.msg + "\n---")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment